mxm commented on code in PR #17553:
URL: https://github.com/apache/iceberg/pull/17553#discussion_r3849530484
##########
spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java:
##########
Review Comment:
Should this file be removed and only be included in the backport for Spark
4.0?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
+ .as("Expected column %s to be shredded with a typed_value
subtree", column)
+ .isTrue();
+ }
+ }
+ }
+
+ private List<MessageType> dataFileSchemas(Table table) throws IOException {
table.refresh();
+ List<MessageType> schemas = Lists.newArrayList();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
assertThat(tasks).isNotEmpty();
+ for (FileScanTask task : tasks) {
+ HadoopInputFile inputFile =
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(task.file().location()), new
Configuration());
+ try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+ schemas.add(reader.getFileMetaData().getSchema());
+ }
+ }
+ }
+ return schemas;
+ }
- FileScanTask task = tasks.iterator().next();
- String path = task.file().location();
-
- HadoopInputFile inputFile =
- HadoopInputFile.fromPath(new org.apache.hadoop.fs.Path(path), new
Configuration());
-
- try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
- MessageType actualSchema = reader.getFileMetaData().getSchema();
- assertThat(actualSchema).isEqualTo(expectedSchema);
+ private static boolean containsTypedValue(Type type) {
+ if (type.isPrimitive()) {
+ return false;
+ }
+ for (Type child : type.asGroupType().getFields()) {
+ if (child.getName().equals("typed_value") || containsTypedValue(child)) {
Review Comment:
Is there a constant for `"typed_value"`?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
+ .as("Expected column %s to be shredded with a typed_value
subtree", column)
+ .isTrue();
+ }
+ }
+ }
+
+ private List<MessageType> dataFileSchemas(Table table) throws IOException {
table.refresh();
+ List<MessageType> schemas = Lists.newArrayList();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
assertThat(tasks).isNotEmpty();
+ for (FileScanTask task : tasks) {
+ HadoopInputFile inputFile =
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(task.file().location()), new
Configuration());
+ try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+ schemas.add(reader.getFileMetaData().getSchema());
+ }
+ }
+ }
+ return schemas;
+ }
- FileScanTask task = tasks.iterator().next();
- String path = task.file().location();
-
- HadoopInputFile inputFile =
- HadoopInputFile.fromPath(new org.apache.hadoop.fs.Path(path), new
Configuration());
-
- try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
- MessageType actualSchema = reader.getFileMetaData().getSchema();
- assertThat(actualSchema).isEqualTo(expectedSchema);
+ private static boolean containsTypedValue(Type type) {
+ if (type.isPrimitive()) {
+ return false;
+ }
+ for (Type child : type.asGroupType().getFields()) {
Review Comment:
NIT: newline after line 967.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
+ .as("Expected column %s to be shredded with a typed_value
subtree", column)
+ .isTrue();
+ }
+ }
+ }
+
+ private List<MessageType> dataFileSchemas(Table table) throws IOException {
table.refresh();
+ List<MessageType> schemas = Lists.newArrayList();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
assertThat(tasks).isNotEmpty();
+ for (FileScanTask task : tasks) {
+ HadoopInputFile inputFile =
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(task.file().location()), new
Configuration());
+ try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+ schemas.add(reader.getFileMetaData().getSchema());
+ }
+ }
+ }
+ return schemas;
Review Comment:
NIT: newline after 960.
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
+ .as("Expected column %s to be shredded with a typed_value
subtree", column)
+ .isTrue();
+ }
+ }
+ }
+
+ private List<MessageType> dataFileSchemas(Table table) throws IOException {
table.refresh();
+ List<MessageType> schemas = Lists.newArrayList();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
assertThat(tasks).isNotEmpty();
+ for (FileScanTask task : tasks) {
+ HadoopInputFile inputFile =
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(task.file().location()), new
Configuration());
+ try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+ schemas.add(reader.getFileMetaData().getSchema());
+ }
+ }
+ }
+ return schemas;
+ }
- FileScanTask task = tasks.iterator().next();
- String path = task.file().location();
-
- HadoopInputFile inputFile =
- HadoopInputFile.fromPath(new org.apache.hadoop.fs.Path(path), new
Configuration());
-
- try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
- MessageType actualSchema = reader.getFileMetaData().getSchema();
- assertThat(actualSchema).isEqualTo(expectedSchema);
+ private static boolean containsTypedValue(Type type) {
+ if (type.isPrimitive()) {
+ return false;
+ }
+ for (Type child : type.asGroupType().getFields()) {
+ if (child.getName().equals("typed_value") || containsTypedValue(child)) {
+ return true;
}
}
+ return false;
Review Comment:
NIT newline after line 972.
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java:
##########
@@ -478,37 +434,96 @@ public void testReadShreddedWithMetricsDisabled(String
metricsMode)
+ "'write.metadata.metrics.default'='%s')",
noStatsTable, metricsMode);
- spark.conf().set("spark.sql.iceberg.shred-variants", "true");
- try {
- sql(
- "INSERT INTO %s VALUES "
- + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), "
- + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))",
- noStatsTable);
- } finally {
- spark.conf().unset("spark.sql.iceberg.shred-variants");
- }
+ insertShredded(
+ "INSERT INTO %s VALUES "
+ + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), "
+ + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))",
+ noStatsTable);
Table table = Spark3Util.loadIcebergTable(spark, noStatsTable);
assertHasTypedValueSubtree(table);
setVectorization(noStatsTable, true);
List<Row> rows = spark.table(noStatsTable).select("id",
"v").orderBy("id").collectAsList();
assertThat(rows).hasSize(2);
- Variant v1 =
- new Variant(
- ((VariantVal) rows.get(0).get(1)).getValue(),
- ((VariantVal) rows.get(0).get(1)).getMetadata());
+ Variant v1 = asVariant(rows.get(0), 1);
assertThat(v1.getFieldByKey("name").getString()).isEqualTo("alice");
- Variant v2 =
- new Variant(
- ((VariantVal) rows.get(1).get(1)).getValue(),
- ((VariantVal) rows.get(1).get(1)).getMetadata());
+ Variant v2 = asVariant(rows.get(1), 1);
assertThat(v2.getFieldByKey("name").getString()).isEqualTo("bob");
sql("DROP TABLE IF EXISTS %s", noStatsTable);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testMultipleUnshreddedVariantColumns(boolean vectorized)
+ throws IOException, NoSuchTableException, ParseException {
+ assertMultipleVariantColumns(false, vectorized);
Review Comment:
Either add a method like `assertMultipleVariantsUnshredded` or add a comment
like this to make this obvious:
```suggestion
assertMultipleVariantColumns(false /* shredded */, vectorized);
```
##########
spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/sql/TestSparkVariantRead.java:
##########
@@ -478,37 +434,96 @@ public void testReadShreddedWithMetricsDisabled(String
metricsMode)
+ "'write.metadata.metrics.default'='%s')",
noStatsTable, metricsMode);
- spark.conf().set("spark.sql.iceberg.shred-variants", "true");
- try {
- sql(
- "INSERT INTO %s VALUES "
- + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), "
- + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))",
- noStatsTable);
- } finally {
- spark.conf().unset("spark.sql.iceberg.shred-variants");
- }
+ insertShredded(
+ "INSERT INTO %s VALUES "
+ + "(1, parse_json('{\"name\":\"alice\",\"age\":30}')), "
+ + "(2, parse_json('{\"name\":\"bob\",\"age\":25}'))",
+ noStatsTable);
Table table = Spark3Util.loadIcebergTable(spark, noStatsTable);
assertHasTypedValueSubtree(table);
setVectorization(noStatsTable, true);
List<Row> rows = spark.table(noStatsTable).select("id",
"v").orderBy("id").collectAsList();
assertThat(rows).hasSize(2);
- Variant v1 =
- new Variant(
- ((VariantVal) rows.get(0).get(1)).getValue(),
- ((VariantVal) rows.get(0).get(1)).getMetadata());
+ Variant v1 = asVariant(rows.get(0), 1);
assertThat(v1.getFieldByKey("name").getString()).isEqualTo("alice");
- Variant v2 =
- new Variant(
- ((VariantVal) rows.get(1).get(1)).getValue(),
- ((VariantVal) rows.get(1).get(1)).getMetadata());
+ Variant v2 = asVariant(rows.get(1), 1);
assertThat(v2.getFieldByKey("name").getString()).isEqualTo("bob");
sql("DROP TABLE IF EXISTS %s", noStatsTable);
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testMultipleUnshreddedVariantColumns(boolean vectorized)
+ throws IOException, NoSuchTableException, ParseException {
+ assertMultipleVariantColumns(false, vectorized);
+ }
+
+ @Test
+ public void testMultipleShreddedVariantColumns()
+ throws IOException, NoSuchTableException, ParseException {
+ assertMultipleVariantColumns(true, true);
Review Comment:
Same here regarding the booleans:
```suggestion
assertMultipleVariantColumns(true /* shredded */, true /* vectorized */);
```
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantWriters.java:
##########
@@ -261,42 +271,157 @@ public void testMixedShredding(Variant variant) throws
IOException {
}
}
+ @Test
+ public void testPartialShreddingWithShreddedObject() throws IOException {
+ VariantMetadata metadata = Variants.metadata("id", "name", "city");
+
+ List<Record> records = Lists.newArrayList();
+ for (int i = 0; i < 3; i++) {
+ ShreddedObject obj = Variants.object(metadata);
+ obj.put("id", Variants.of(1000L + i));
+ obj.put("name", Variants.of("user_" + i));
+ obj.put("city", Variants.of("city_" + i));
+
+ Variant variant = Variant.of(metadata, obj);
+ Record record = RECORD.copy("id", i, "var", variant);
+ records.add(record);
+ }
+
+ VariantShreddingFunction partialShredding = (id, name) -> shredOnly("id");
+
+ List<Record> actual = writeAndRead(SCHEMA, partialShredding, records,
"var");
+
+ assertThat(actual).hasSameSizeAs(records);
+ for (int i = 0; i < records.size(); i++) {
+ Record expected = records.get(i);
+ Record read = actual.get(i);
+
+ InternalTestHelpers.assertEquals(SCHEMA.asStruct(), expected, read);
+
+ Variant readVariant = (Variant) read.getField("var");
+ VariantObject readObj = readVariant.value().asObject();
+ assertThat(readObj.numFields()).isEqualTo(3);
+ assertThat(readObj.get("id").asPrimitive().get()).isEqualTo(1000L + i);
+ assertThat(readObj.get("name").asPrimitive().get()).isEqualTo("user_" +
i);
+ assertThat(readObj.get("city").asPrimitive().get()).isEqualTo("city_" +
i);
+ }
+ }
+
+ @Test
+ public void testPartialShreddingMultipleColumns() throws IOException {
+ VariantMetadata metadata1 = Variants.metadata("id", "name", "city");
+ VariantMetadata metadata2 = Variants.metadata("key", "val", "extra");
+
+ List<Record> records = Lists.newArrayList();
+ for (int i = 0; i < 3; i++) {
+ ShreddedObject object1 = Variants.object(metadata1);
+ object1.put("id", Variants.of(1000L + i));
+ object1.put("name", Variants.of("user_" + i));
+ object1.put("city", Variants.of("city_" + i));
+
+ ShreddedObject object2 = Variants.object(metadata2);
+ object2.put("key", Variants.of(2000L + i));
+ object2.put("val", Variants.of("val_" + i));
+ object2.put("extra", Variants.of("extra_" + i));
+
+ records.add(
+ RECORD_TWO_VARIANTS.copy(
+ "id", i,
+ "var1", Variant.of(metadata1, object1),
+ "var2", Variant.of(metadata2, object2)));
+ }
Review Comment:
In tests, I'd prefer explicit construction of test data over loops. Why
three iterations here?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
Review Comment:
Why asserting only the first?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
+ .as("Expected column %s to be shredded with a typed_value
subtree", column)
+ .isTrue();
+ }
+ }
+ }
+
+ private List<MessageType> dataFileSchemas(Table table) throws IOException {
table.refresh();
+ List<MessageType> schemas = Lists.newArrayList();
try (CloseableIterable<FileScanTask> tasks = table.newScan().planFiles()) {
assertThat(tasks).isNotEmpty();
+ for (FileScanTask task : tasks) {
+ HadoopInputFile inputFile =
+ HadoopInputFile.fromPath(
+ new org.apache.hadoop.fs.Path(task.file().location()), new
Configuration());
+ try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
+ schemas.add(reader.getFileMetaData().getSchema());
+ }
+ }
+ }
+ return schemas;
+ }
- FileScanTask task = tasks.iterator().next();
- String path = task.file().location();
-
- HadoopInputFile inputFile =
- HadoopInputFile.fromPath(new org.apache.hadoop.fs.Path(path), new
Configuration());
-
- try (ParquetFileReader reader = ParquetFileReader.open(inputFile)) {
- MessageType actualSchema = reader.getFileMetaData().getSchema();
- assertThat(actualSchema).isEqualTo(expectedSchema);
+ private static boolean containsTypedValue(Type type) {
+ if (type.isPrimitive()) {
+ return false;
+ }
+ for (Type child : type.asGroupType().getFields()) {
+ if (child.getName().equals("typed_value") || containsTypedValue(child)) {
+ return true;
}
}
+ return false;
}
private static MessageType parquetSchema(Type variantTypes) {
Review Comment:
Should we change this to:
```suggestion
private static MessageType parquetSchema(Type.. variantTypes) {
```
We can then assert using the pattern of the other tests, e.g.:
```java
GroupType v1 = ...
GroupType v2 = ...
MessageType expectedSchema = parquetSchema(v1, v2);
verifyParquetSchema(icebergTable, expectedSchema);
```
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantWriters.java:
##########
@@ -311,52 +436,21 @@ private static ValueArray array(VariantValue... values) {
return arr;
}
- @Test
- public void testPartialShreddingWithShreddedObject() throws IOException {
- // Test for issue #15086: partial shredding with ShreddedObject created
using put()
- // Create a ShreddedObject with multiple fields, then partially shred it
Review Comment:
Should this comment be deleted?
##########
flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/TestFlinkVariantShreddingType.java:
##########
@@ -876,22 +876,101 @@ public void testDecimalFallbackAfterBuffer() throws
IOException {
sql("DROP TEMPORARY VIEW IF EXISTS tmp_source");
}
+ @TestTemplate
+ public void testReadMultipleShreddedVariantColumns() throws IOException {
+ String multiTable = "multi_shredded_variant";
+ sql(
+ """
+ CREATE TABLE %s (
+ id int NOT NULL,
+ v1 variant NOT NULL,
+ v2 variant NOT NULL
+ ) WITH (
+ 'write.format.default'='%s',
+ 'format-version'='3',
+ 'shred-variants'='true',
+ 'variant-inference-buffer-size'='10'
+ )
+ """,
+ multiTable, FileFormat.PARQUET.name());
+
+ sql(
+ """
+ INSERT INTO %s VALUES
+ (1, parse_json('{"name": "Alice", "age": 30}'),
parse_json('{"city": "Seattle", "zip": 98101}')),
+ (2, parse_json('{"name": "Bob", "age": 25}'),
parse_json('{"city": "Portland", "zip": 97201}'))
+ """,
+ multiTable);
+
+ Table table =
validationCatalog.loadTable(TableIdentifier.of(icebergNamespace, multiTable));
+ assertShredded(table, "v1", "v2");
+
+ List<Row> rows =
+ sql(
+ """
+ SELECT id,
+ json_value(v1, '$.name'),
+ json_value(v1, '$.age' RETURNING int),
+ json_value(v2, '$.city'),
+ json_value(v2, '$.zip' RETURNING int)
+ FROM %s
+ ORDER BY id
+ """,
+ multiTable);
+ assertThat(rows).hasSize(2);
+ assertThat(rows.get(0).getField(1)).isEqualTo("Alice");
+ assertThat(rows.get(0).getField(2)).isEqualTo(30);
+ assertThat(rows.get(0).getField(3)).isEqualTo("Seattle");
+ assertThat(rows.get(0).getField(4)).isEqualTo(98101);
+ assertThat(rows.get(1).getField(1)).isEqualTo("Bob");
+ assertThat(rows.get(1).getField(2)).isEqualTo(25);
+ assertThat(rows.get(1).getField(3)).isEqualTo("Portland");
+ assertThat(rows.get(1).getField(4)).isEqualTo(97201);
+
+ sql("DROP TABLE %s", multiTable);
+ }
+
private void verifyParquetSchema(Table table, MessageType expectedSchema)
throws IOException {
+ assertThat(dataFileSchemas(table).get(0)).isEqualTo(expectedSchema);
+ }
+
+ private void assertShredded(Table table, String... columns) throws
IOException {
+ for (MessageType schema : dataFileSchemas(table)) {
+ for (String column : columns) {
+ assertThat(containsTypedValue(schema.getType(column)))
Review Comment:
This only checks that the type of the field exists but not the field name.
Can we verify the name?
##########
parquet/src/test/java/org/apache/iceberg/parquet/TestVariantWriters.java:
##########
@@ -261,42 +271,157 @@ public void testMixedShredding(Variant variant) throws
IOException {
}
}
+ @Test
+ public void testPartialShreddingWithShreddedObject() throws IOException {
+ VariantMetadata metadata = Variants.metadata("id", "name", "city");
+
+ List<Record> records = Lists.newArrayList();
+ for (int i = 0; i < 3; i++) {
+ ShreddedObject obj = Variants.object(metadata);
+ obj.put("id", Variants.of(1000L + i));
+ obj.put("name", Variants.of("user_" + i));
+ obj.put("city", Variants.of("city_" + i));
+
+ Variant variant = Variant.of(metadata, obj);
+ Record record = RECORD.copy("id", i, "var", variant);
+ records.add(record);
+ }
+
+ VariantShreddingFunction partialShredding = (id, name) -> shredOnly("id");
+
+ List<Record> actual = writeAndRead(SCHEMA, partialShredding, records,
"var");
+
+ assertThat(actual).hasSameSizeAs(records);
+ for (int i = 0; i < records.size(); i++) {
+ Record expected = records.get(i);
+ Record read = actual.get(i);
+
+ InternalTestHelpers.assertEquals(SCHEMA.asStruct(), expected, read);
+
+ Variant readVariant = (Variant) read.getField("var");
+ VariantObject readObj = readVariant.value().asObject();
+ assertThat(readObj.numFields()).isEqualTo(3);
+ assertThat(readObj.get("id").asPrimitive().get()).isEqualTo(1000L + i);
+ assertThat(readObj.get("name").asPrimitive().get()).isEqualTo("user_" +
i);
+ assertThat(readObj.get("city").asPrimitive().get()).isEqualTo("city_" +
i);
+ }
+ }
+
+ @Test
+ public void testPartialShreddingMultipleColumns() throws IOException {
+ VariantMetadata metadata1 = Variants.metadata("id", "name", "city");
+ VariantMetadata metadata2 = Variants.metadata("key", "val", "extra");
+
+ List<Record> records = Lists.newArrayList();
+ for (int i = 0; i < 3; i++) {
+ ShreddedObject object1 = Variants.object(metadata1);
+ object1.put("id", Variants.of(1000L + i));
+ object1.put("name", Variants.of("user_" + i));
+ object1.put("city", Variants.of("city_" + i));
+
+ ShreddedObject object2 = Variants.object(metadata2);
+ object2.put("key", Variants.of(2000L + i));
+ object2.put("val", Variants.of("val_" + i));
+ object2.put("extra", Variants.of("extra_" + i));
+
+ records.add(
+ RECORD_TWO_VARIANTS.copy(
+ "id", i,
+ "var1", Variant.of(metadata1, object1),
+ "var2", Variant.of(metadata2, object2)));
+ }
+
+ VariantShreddingFunction partialShredding =
+ (id, name) -> {
+ if (name.equals("var1")) {
+ return shredOnly("id");
+ } else if (name.equals("var2")) {
+ return shredOnly("key");
+ }
+ return null;
+ };
+
+ List<Record> actual =
+ writeAndRead(SCHEMA_TWO_VARIANTS, partialShredding, records, "var1",
"var2");
+
+ assertThat(actual).hasSameSizeAs(records);
+ for (int i = 0; i < records.size(); i++) {
+ InternalTestHelpers.assertEquals(
+ SCHEMA_TWO_VARIANTS.asStruct(), records.get(i), actual.get(i));
+
+ VariantObject readObject1 = ((Variant)
actual.get(i).getField("var1")).value().asObject();
+ assertThat(readObject1.numFields()).isEqualTo(3);
+ assertThat(readObject1.get("id").asPrimitive().get()).isEqualTo(1000L +
i);
+
assertThat(readObject1.get("name").asPrimitive().get()).isEqualTo("user_" + i);
+
assertThat(readObject1.get("city").asPrimitive().get()).isEqualTo("city_" + i);
+
+ VariantObject readObject2 = ((Variant)
actual.get(i).getField("var2")).value().asObject();
+ assertThat(readObject2.numFields()).isEqualTo(3);
+ assertThat(readObject2.get("key").asPrimitive().get()).isEqualTo(2000L +
i);
+ assertThat(readObject2.get("val").asPrimitive().get()).isEqualTo("val_"
+ i);
+
assertThat(readObject2.get("extra").asPrimitive().get()).isEqualTo("extra_" +
i);
+ }
+ }
Review Comment:
Same here, I'd prefer not to use loops for assertions. Makes test failures
easier to reason about.
--
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]
