vbhanuchander-lang commented on code in PR #17644:
URL: https://github.com/apache/iceberg/pull/17644#discussion_r4107503570
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/AvroWithFlinkSchemaVisitor.java:
##########
@@ -51,12 +55,21 @@ protected LogicalType arrayElementType(LogicalType
arrayType) {
@Override
protected LogicalType mapKeyType(LogicalType mapType) {
Preconditions.checkArgument(isMapType(mapType), "Invalid map: %s is not a
map", mapType);
+ if (mapType instanceof MultisetType) {
+ return ((MultisetType) mapType).getElementType();
+ }
+
return ((MapType) mapType).getKeyType();
}
@Override
protected LogicalType mapValueType(LogicalType mapType) {
Preconditions.checkArgument(isMapType(mapType), "Invalid map: %s is not a
map", mapType);
+ if (mapType instanceof MultisetType) {
+ // the occurrence count is a required int
+ return new IntType(false);
Review Comment:
Good catch. Both work, so it is about which describes the type correctly
rather than a bug. `FlinkTypeToType#visit(MultisetType)` produces
`Types.MapType.ofRequired(...)`, so the Iceberg schema the writers must match
has a **required** count, which is why I used `new IntType(false)`. I checked
whether it is load-bearing by flipping it to `new IntType()` and re-running:
all six tests still pass, because the writers use the logical type only to
choose a value writer and a multiset count is never null. So
`AvroSchemaConverter`'s `new IntType()` is looser than the schema rather than
wrong; I have left it alone since it is a read-side conversion, but happy to
tighten it in a follow-up if you would rather the two agreed.
##########
flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/data/ParquetWithFlinkSchemaVisitor.java:
##########
@@ -220,4 +227,13 @@ protected String[] path(String name) {
list.add(name);
return list.toArray(new String[0]);
}
+
+ private static MapType toMapType(LogicalType sType) {
+ if (sType instanceof MultisetType) {
+ return new MapType(
+ sType.isNullable(), ((MultisetType) sType).getElementType(), new
IntType(false));
+ }
Review Comment:
Adopted — `FlinkMultisets#asMapType` uses `if (logicalType instanceof
MultisetType multisetType)`. The module compiles at `options.release = 17` so
it is available. The method moved out of `ParquetWithFlinkSchemaVisitor` into
the shared `FlinkMultisets`, since @pvary asked for the duplication across the
three visitors to be extracted.
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkMultisetWrite.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.data;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+import java.util.Map;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+/**
+ * A Flink {@code MULTISET<T>} is converted to an Iceberg {@code map<T, int>}
of element to
+ * occurrence count by {@code FlinkTypeToType#visit(MultisetType)}, so a table
with a multiset
+ * column can be created. The write path pairs the Iceberg map with the Flink
{@link MultisetType},
+ * which is not a {@code MapType}, so every data file writer used to reject it
and the column could
+ * never be written.
+ *
+ * <p>The existing writer tests cannot cover this: they derive the Flink type
with {@code
+ * FlinkSchemaUtil.convert(icebergSchema)}, which turns {@code map<string,
int>} back into a {@code
+ * MapType} and never produces a {@link MultisetType}.
+ */
+public class TestFlinkMultisetWrite {
+
+ private static final Schema SCHEMA =
+ new Schema(
+ Types.NestedField.required(1, "id", Types.IntegerType.get()),
+ Types.NestedField.optional(
+ 2,
+ "tags",
+ Types.MapType.ofRequired(3, 4, Types.StringType.get(),
Types.IntegerType.get())));
+
+ // ROW<id INT NOT NULL, tags MULTISET<STRING NOT NULL>> — what Flink hands
the writers for a
+ // multiset column, and what FlinkSchemaUtil.convert(SCHEMA) can never
produce.
+ private static final RowType FLINK_TYPE =
+ RowType.of(
+ new LogicalType[] {
+ new IntType(false),
+ new MultisetType(true, new VarCharType(false,
VarCharType.MAX_LENGTH))
+ },
+ new String[] {"id", "tags"});
+
+ private static RowData row() {
+ Map<Object, Object> counts =
+ ImmutableMap.of(StringData.fromString("a"), 2,
StringData.fromString("b"), 1);
+ return GenericRowData.of(1, new GenericMapData(counts));
+ }
+
+ @Test
+ public void testParquetAcceptsMultiset() {
+ assertThatCode(
+ () -> {
+ OutputFile out = new InMemoryOutputFile();
+ try (FileAppender<RowData> writer =
+ Parquet.write(out)
+ .schema(SCHEMA)
+ .createWriterFunc(
+ msgType ->
FlinkParquetWriters.buildWriter(FLINK_TYPE, msgType))
+ .build()) {
+ writer.add(row());
+ }
+ })
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testAvroAcceptsMultiset() {
+ assertThatCode(
+ () -> {
+ OutputFile out = new InMemoryOutputFile();
+ try (FileAppender<RowData> writer =
+ Avro.write(out)
+ .schema(SCHEMA)
+ .createWriterFunc(ignored -> new
FlinkAvroWriter(FLINK_TYPE))
+ .build()) {
+ writer.add(row());
+ }
+ })
+ .doesNotThrowAnyException();
+ }
+
+ @Test
+ public void testOrcAcceptsMultiset() {
+ assertThatCode(
+ () -> {
+ java.io.File orcFile =
+ new java.io.File(
+
java.nio.file.Files.createTempDirectory("multiset").toFile(), "data.orc");
Review Comment:
Fixed — the ORC test uses `@TempDir` with the `File.createTempFile("junit",
null, temp.toFile())` idiom the other tests in this package use, so nothing is
left behind. The Parquet and Avro tests need no file at all now:
`InMemoryOutputFile.toInputFile()` reads straight back.
##########
flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/data/TestFlinkMultisetWrite.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.iceberg.flink.data;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+import java.util.Map;
+import org.apache.flink.table.data.GenericMapData;
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.flink.table.types.logical.IntType;
+import org.apache.flink.table.types.logical.LogicalType;
+import org.apache.flink.table.types.logical.MultisetType;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.inmemory.InMemoryOutputFile;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.io.OutputFile;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.parquet.Parquet;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.jupiter.api.Test;
+
+/**
+ * A Flink {@code MULTISET<T>} is converted to an Iceberg {@code map<T, int>}
of element to
+ * occurrence count by {@code FlinkTypeToType#visit(MultisetType)}, so a table
with a multiset
+ * column can be created. The write path pairs the Iceberg map with the Flink
{@link MultisetType},
+ * which is not a {@code MapType}, so every data file writer used to reject it
and the column could
+ * never be written.
+ *
+ * <p>The existing writer tests cannot cover this: they derive the Flink type
with {@code
+ * FlinkSchemaUtil.convert(icebergSchema)}, which turns {@code map<string,
int>} back into a {@code
+ * MapType} and never produces a {@link MultisetType}.
+ */
+public class TestFlinkMultisetWrite {
Review Comment:
Added — all five write tests are round trips now, writing and then reading
back through the matching reader and comparing `id` and the tag counts, across
Parquet, Avro and ORC.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
