You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by ao...@apache.org on 2021/09/15 00:43:59 UTC
[iceberg] branch master updated: Core: Promote sorted column
metrics to truncate(16) (#2240)
This is an automated email from the ASF dual-hosted git repository.
aokolnychyi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/master by this push:
new 5d1aa35 Core: Promote sorted column metrics to truncate(16) (#2240)
5d1aa35 is described below
commit 5d1aa357509c6f42067dbb0e21c76094ab4c8081
Author: Szehon Ho <sz...@gmail.com>
AuthorDate: Tue Sep 14 17:43:53 2021 -0700
Core: Promote sorted column metrics to truncate(16) (#2240)
---
.../java/org/apache/iceberg/MetricsConfig.java | 38 ++++++
.../main/java/org/apache/iceberg/avro/Avro.java | 3 +-
.../org/apache/iceberg/util/SortOrderUtil.java | 17 +++
.../test/java/org/apache/iceberg/TestMetrics.java | 110 +++++++++++++++
.../java/org/apache/iceberg/TestMetricsModes.java | 85 ++++++++++++
.../java/org/apache/iceberg/TestSortOrder.java | 26 ++++
.../org/apache/iceberg/data/BaseWriterFactory.java | 4 +-
.../org/apache/iceberg/io/TestWriterMetrics.java | 149 +++++++++++++++++++++
.../org/apache/iceberg/orc/TestOrcMetrics.java | 17 ++-
.../apache/iceberg/parquet/TestParquetMetrics.java | 15 ++-
.../iceberg/flink/sink/TestFlinkWriterMetrics.java | 51 +++++++
orc/src/main/java/org/apache/iceberg/orc/ORC.java | 2 +-
.../java/org/apache/iceberg/parquet/Parquet.java | 6 +-
.../org/apache/iceberg/spark/SparkTableUtil.java | 2 +-
.../spark/source/TestSparkWriterMetrics.java | 58 ++++++++
15 files changed, 566 insertions(+), 17 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/MetricsConfig.java b/core/src/main/java/org/apache/iceberg/MetricsConfig.java
index be76cf6..4d0aef8 100644
--- a/core/src/main/java/org/apache/iceberg/MetricsConfig.java
+++ b/core/src/main/java/org/apache/iceberg/MetricsConfig.java
@@ -22,9 +22,11 @@ package org.apache.iceberg;
import java.io.Serializable;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import org.apache.iceberg.MetricsModes.MetricsMode;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+import org.apache.iceberg.util.SortOrderUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -76,7 +78,25 @@ public class MetricsConfig implements Serializable {
}
}
+ /**
+ * Creates a metrics config from table configuration.
+ * @param props table configuration
+ * @deprecated use {@link MetricsConfig#forTable(Table)}
+ **/
+ @Deprecated
public static MetricsConfig fromProperties(Map<String, String> props) {
+ return from(props, null);
+ }
+
+ /**
+ * Creates a metrics config from a table.
+ * @param table iceberg table
+ */
+ public static MetricsConfig forTable(Table table) {
+ return from(table.properties(), table.sortOrder());
+ }
+
+ private static MetricsConfig from(Map<String, String> props, SortOrder order) {
MetricsConfig spec = new MetricsConfig();
String defaultModeAsString = props.getOrDefault(DEFAULT_WRITE_METRICS_MODE, DEFAULT_WRITE_METRICS_MODE_DEFAULT);
try {
@@ -87,6 +107,11 @@ public class MetricsConfig implements Serializable {
spec.defaultMode = MetricsModes.fromString(DEFAULT_WRITE_METRICS_MODE_DEFAULT);
}
+ // First set sorted column with sorted column default (can be overridden by user)
+ MetricsMode sortedColDefaultMode = sortedColumnDefaultMode(spec.defaultMode);
+ Set<String> sortedCols = SortOrderUtil.orderPreservingSortedColumns(order);
+ sortedCols.forEach(sc -> spec.columnModes.put(sc, sortedColDefaultMode));
+
props.keySet().stream()
.filter(key -> key.startsWith(METRICS_MODE_COLUMN_CONF_PREFIX))
.forEach(key -> {
@@ -105,6 +130,19 @@ public class MetricsConfig implements Serializable {
return spec;
}
+ /**
+ * Auto promote sorted columns to truncate(16) if default is set at Counts or None.
+ * @param defaultMode default mode
+ * @return mode to use
+ */
+ private static MetricsMode sortedColumnDefaultMode(MetricsMode defaultMode) {
+ if (defaultMode == MetricsModes.None.get() || defaultMode == MetricsModes.Counts.get()) {
+ return MetricsModes.Truncate.withLength(16);
+ } else {
+ return defaultMode;
+ }
+ }
+
public void validateReferencedColumns(Schema schema) {
for (String column : columnModes.keySet()) {
ValidationException.check(
diff --git a/core/src/main/java/org/apache/iceberg/avro/Avro.java b/core/src/main/java/org/apache/iceberg/avro/Avro.java
index 59715b0..9255597 100644
--- a/core/src/main/java/org/apache/iceberg/avro/Avro.java
+++ b/core/src/main/java/org/apache/iceberg/avro/Avro.java
@@ -118,6 +118,7 @@ public class Avro {
public WriteBuilder forTable(Table table) {
schema(table.schema());
setAll(table.properties());
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
@@ -256,7 +257,7 @@ public class Avro {
schema(table.schema());
withSpec(table.spec());
setAll(table.properties());
- metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
diff --git a/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java b/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
index 9cd0ff9..54941e1 100644
--- a/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
+++ b/core/src/main/java/org/apache/iceberg/util/SortOrderUtil.java
@@ -20,6 +20,10 @@
package org.apache.iceberg.util;
import java.util.Collection;
+import java.util.Collections;
+import java.util.Objects;
+import java.util.Set;
+import java.util.stream.Collectors;
import org.apache.iceberg.PartitionField;
import org.apache.iceberg.PartitionSpec;
import org.apache.iceberg.Schema;
@@ -64,4 +68,17 @@ public class SortOrderUtil {
return builder.build();
}
+
+ public static Set<String> orderPreservingSortedColumns(SortOrder sortOrder) {
+ if (sortOrder == null) {
+ return Collections.emptySet();
+ } else {
+ return sortOrder.fields().stream()
+ .filter(f -> f.transform().preservesOrder())
+ .map(SortField::sourceId)
+ .map(sid -> sortOrder.schema().findColumnName(sid))
+ .filter(Objects::nonNull)
+ .collect(Collectors.toSet());
+ }
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestMetrics.java b/core/src/test/java/org/apache/iceberg/TestMetrics.java
index e00dcab..b8d2c4d 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetrics.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetrics.java
@@ -19,6 +19,7 @@
package org.apache.iceberg;
+import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.nio.ByteBuffer;
@@ -53,9 +54,12 @@ import org.apache.iceberg.types.Types.StructType;
import org.apache.iceberg.types.Types.TimeType;
import org.apache.iceberg.types.Types.TimestampType;
import org.apache.iceberg.util.DateTimeUtil;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Assume;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
import static org.apache.iceberg.types.Conversions.fromByteBuffer;
import static org.apache.iceberg.types.Types.NestedField.optional;
@@ -66,6 +70,13 @@ import static org.apache.iceberg.types.Types.NestedField.required;
*/
public abstract class TestMetrics {
+ protected TestMetrics(int formatVersion) {
+ this.formatVersion = formatVersion;
+ }
+
+ @Rule
+ public TemporaryFolder temp = new TemporaryFolder();
+
private static final StructType LEAF_STRUCT_TYPE = StructType.of(
optional(5, "leafLongCol", LongType.get()),
optional(6, "leafBinaryCol", BinaryType.get())
@@ -107,8 +118,14 @@ public abstract class TestMetrics {
private static final Record FLOAT_DOUBLE_RECORD_2 = createRecordWithFloatAndDouble(5.6F, 7.8D);
private static final Record NAN_ONLY_RECORD = createRecordWithFloatAndDouble(Float.NaN, Double.NaN);
+ private final int formatVersion;
private final byte[] fixed = "abcd".getBytes(StandardCharsets.UTF_8);
+ @After
+ public void after() {
+ TestTables.clearTables();
+ }
+
private static Record createRecordWithFloatAndDouble(float floatValue, double doubleValue) {
Record record = GenericRecord.create(FLOAT_DOUBLE_ONLY_SCHEMA);
record.setField("floatCol", floatValue);
@@ -613,6 +630,99 @@ public abstract class TestMetrics {
assertBounds(1, Types.BinaryType.get(), expectedMinBounds, expectedMaxBounds, metrics);
}
+ @Test
+ public void testSortedColumnMetrics() throws IOException {
+ File tableDir = temp.newFolder();
+ tableDir.delete(); // created by table create
+
+ SortOrder sortOrder = SortOrder.builderFor(SIMPLE_SCHEMA)
+ .asc("booleanCol")
+ .asc("intCol")
+ .asc("longCol")
+ .asc("decimalCol")
+ .asc("stringCol")
+ .asc("dateCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = TestTables.create(tableDir, "test", SIMPLE_SCHEMA, spec, sortOrder, formatVersion);
+ table.updateProperties().set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none").commit();
+
+ Record firstRecord = GenericRecord.create(SIMPLE_SCHEMA);
+ firstRecord.setField("booleanCol", true);
+ firstRecord.setField("intCol", Integer.MIN_VALUE);
+ firstRecord.setField("longCol", Long.MIN_VALUE);
+ firstRecord.setField("floatCol", Float.NaN);
+ firstRecord.setField("doubleCol", 2.0D);
+ firstRecord.setField("decimalCol", new BigDecimal("0.00"));
+ firstRecord.setField("stringCol", "AAA");
+ firstRecord.setField("dateCol", DateTimeUtil.dateFromDays(1500));
+ firstRecord.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
+ firstRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
+ firstRecord.setField("fixedCol", fixed);
+ firstRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
+ firstRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(0L));
+
+ Record secondRecord = GenericRecord.create(SIMPLE_SCHEMA);
+
+ secondRecord.setField("booleanCol", false);
+ secondRecord.setField("intCol", Integer.MAX_VALUE);
+ secondRecord.setField("longCol", Long.MAX_VALUE);
+ secondRecord.setField("floatCol", Float.NaN);
+ secondRecord.setField("doubleCol", 2.0D);
+ secondRecord.setField("decimalCol", new BigDecimal("10.00"));
+ secondRecord.setField("stringCol", "ZZZ");
+ secondRecord.setField("dateCol", DateTimeUtil.dateFromDays(3000));
+ secondRecord.setField("timeCol", DateTimeUtil.timeFromMicros(2000L));
+ secondRecord.setField("timestampColAboveEpoch", DateTimeUtil.timestampFromMicros(0L));
+ secondRecord.setField("fixedCol", fixed);
+ secondRecord.setField("binaryCol", ByteBuffer.wrap("S".getBytes()));
+ secondRecord.setField("timestampColBelowEpoch", DateTimeUtil.timestampFromMicros(0L));
+
+ Metrics metrics = getMetrics(
+ SIMPLE_SCHEMA,
+ MetricsConfig.forTable(table),
+ firstRecord, secondRecord);
+
+ Assert.assertEquals(2L, (long) metrics.recordCount());
+ assertBounds(1, BooleanType.get(), false, true, metrics);
+ assertBounds(2, IntegerType.get(), Integer.MIN_VALUE, Integer.MAX_VALUE, metrics);
+ assertBounds(3, LongType.get(), Long.MIN_VALUE, Long.MAX_VALUE, metrics);
+ assertBounds(6, DecimalType.of(10, 2),
+ new BigDecimal("0.00"), new BigDecimal("10.00"), metrics);
+ assertBounds(7, StringType.get(),
+ CharBuffer.wrap("AAA"), CharBuffer.wrap("ZZZ"), metrics);
+ assertBounds(8, DateType.get(), 1500, 3000, metrics);
+ }
+
+ @Test
+ public void testMetricsForSortedNestedStructFields() throws IOException {
+ File tableDir = temp.newFolder();
+ tableDir.delete(); // created by table create
+
+ SortOrder sortOrder = SortOrder.builderFor(NESTED_SCHEMA)
+ .asc("nestedStructCol.longCol")
+ .asc("nestedStructCol.leafStructCol.leafLongCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
+ Table table = TestTables.create(tableDir, "nested", NESTED_SCHEMA, spec, sortOrder, formatVersion);
+
+ Record leafStruct = GenericRecord.create(LEAF_STRUCT_TYPE);
+ leafStruct.setField("leafLongCol", Long.MAX_VALUE);
+ leafStruct.setField("leafBinaryCol", ByteBuffer.wrap("A".getBytes()));
+ Record nestedStruct = GenericRecord.create(NESTED_STRUCT_TYPE);
+ nestedStruct.setField("longCol", Long.MAX_VALUE);
+ nestedStruct.setField("leafStructCol", leafStruct);
+ nestedStruct.setField("doubleCol", Double.NaN);
+ Record record = GenericRecord.create(NESTED_SCHEMA);
+ record.setField("intCol", Integer.MAX_VALUE);
+ record.setField("nestedStructCol", nestedStruct);
+
+ Metrics metrics = getMetrics(NESTED_SCHEMA,
+ MetricsConfig.forTable(table),
+ record);
+
+ assertBounds(3, LongType.get(), Long.MAX_VALUE, Long.MAX_VALUE, metrics);
+ assertBounds(5, LongType.get(), Long.MAX_VALUE, Long.MAX_VALUE, metrics);
+ }
+
protected void assertCounts(int fieldId, Long valueCount, Long nullValueCount, Metrics metrics) {
assertCounts(fieldId, valueCount, nullValueCount, null, metrics);
}
diff --git a/core/src/test/java/org/apache/iceberg/TestMetricsModes.java b/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
index 65a31d4..8660f81 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetricsModes.java
@@ -19,22 +19,50 @@
package org.apache.iceberg;
+import java.io.File;
import java.util.Map;
import org.apache.iceberg.MetricsModes.Counts;
import org.apache.iceberg.MetricsModes.Full;
import org.apache.iceberg.MetricsModes.None;
import org.apache.iceberg.MetricsModes.Truncate;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+@RunWith(Parameterized.class)
public class TestMetricsModes {
+ private final int formatVersion;
+
+ @Parameterized.Parameters(name = "formatVersion = {0}")
+ public static Object[] parameters() {
+ return new Object[] { 1, 2 };
+ }
+
+ public TestMetricsModes(int formatVersion) {
+ this.formatVersion = formatVersion;
+ }
+
@Rule
public ExpectedException exceptionRule = ExpectedException.none();
+ @Rule
+ public TemporaryFolder temp = new TemporaryFolder();
+
+ @After
+ public void after() {
+ TestTables.clearTables();
+ }
+
@Test
public void testMetricsModeParsing() {
Assert.assertEquals(None.get(), MetricsModes.fromString("none"));
@@ -75,4 +103,61 @@ public class TestMetricsModes {
Assert.assertEquals("Invalid mode should be defaulted to library default (truncate(16))",
MetricsModes.Truncate.withLength(16), config.columnMode("col"));
}
+
+ @Test
+ public void testMetricsConfigSortedColsDefault() throws Exception {
+ File tableDir = temp.newFolder();
+ tableDir.delete(); // created by table create
+
+ Schema schema = new Schema(
+ required(1, "col1", Types.IntegerType.get()),
+ required(2, "col2", Types.IntegerType.get()),
+ required(3, "col3", Types.IntegerType.get()),
+ required(4, "col4", Types.IntegerType.get())
+ );
+ SortOrder sortOrder = SortOrder.builderFor(schema).asc("col2").asc("col3").build();
+ Table testTable = TestTables.create(tableDir, "test", schema, PartitionSpec.unpartitioned(),
+ sortOrder, formatVersion);
+ testTable.updateProperties()
+ .set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "counts")
+ .set(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "col1", "counts")
+ .set(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "col2", "none")
+ .commit();
+
+ MetricsConfig config = MetricsConfig.forTable(testTable);
+ Assert.assertEquals("Non-sorted existing column should not be overridden",
+ Counts.get(), config.columnMode("col1"));
+ Assert.assertEquals("Sorted column defaults should not override user specified config",
+ None.get(), config.columnMode("col2"));
+ Assert.assertEquals("Unspecified sorted column should use default",
+ Truncate.withLength(16), config.columnMode("col3"));
+ Assert.assertEquals("Unspecified normal column should use default",
+ Counts.get(), config.columnMode("col4"));
+ }
+
+ @Test
+ public void testMetricsConfigSortedColsDefaultByInvalid() throws Exception {
+ File tableDir = temp.newFolder();
+ tableDir.delete(); // created by table create
+
+ Schema schema = new Schema(
+ required(1, "col1", Types.IntegerType.get()),
+ required(2, "col2", Types.IntegerType.get()),
+ required(3, "col3", Types.IntegerType.get())
+ );
+ SortOrder sortOrder = SortOrder.builderFor(schema).asc("col2").asc("col3").build();
+ Table testTable = TestTables.create(tableDir, "test", schema, PartitionSpec.unpartitioned(),
+ sortOrder, formatVersion);
+ testTable.updateProperties()
+ .set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "counts")
+ .set(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "col1", "full")
+ .set(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX + "col2", "invalid")
+ .commit();
+
+ MetricsConfig config = MetricsConfig.forTable(testTable);
+ Assert.assertEquals("Non-sorted existing column should not be overridden by sorted column",
+ Full.get(), config.columnMode("col1"));
+ Assert.assertEquals("Original default applies as user entered invalid mode for sorted column",
+ Counts.get(), config.columnMode("col2"));
+ }
}
diff --git a/core/src/test/java/org/apache/iceberg/TestSortOrder.java b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
index d20bbf6..fe99e7a 100644
--- a/core/src/test/java/org/apache/iceberg/TestSortOrder.java
+++ b/core/src/test/java/org/apache/iceberg/TestSortOrder.java
@@ -21,9 +21,12 @@ package org.apache.iceberg;
import java.io.File;
import java.io.IOException;
+import java.util.Set;
import org.apache.iceberg.exceptions.ValidationException;
import org.apache.iceberg.expressions.Expressions;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.types.Types;
+import org.apache.iceberg.util.SortOrderUtil;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -35,6 +38,7 @@ import org.junit.runners.Parameterized;
import static org.apache.iceberg.NullOrder.NULLS_FIRST;
import static org.apache.iceberg.NullOrder.NULLS_LAST;
+import static org.apache.iceberg.expressions.Expressions.bucket;
import static org.apache.iceberg.expressions.Expressions.truncate;
import static org.apache.iceberg.types.Types.NestedField.optional;
import static org.apache.iceberg.types.Types.NestedField.required;
@@ -315,4 +319,26 @@ public class TestSortOrder {
SortOrder order = SortOrder.builderFor(SCHEMA).build();
Assert.assertEquals("Order must be unsorted", SortOrder.unsorted(), order);
}
+
+ @Test
+ public void testSortedColumnNames() {
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(10)
+ .asc("s.id")
+ .desc(truncate("data", 10))
+ .build();
+ Set<String> sortedCols = SortOrderUtil.orderPreservingSortedColumns(order);
+ Assert.assertEquals(ImmutableSet.of("s.id", "data"), sortedCols);
+ }
+
+ @Test
+ public void testPreservingOrderSortedColumnNames() {
+ SortOrder order = SortOrder.builderFor(SCHEMA)
+ .withOrderId(10)
+ .asc(bucket("s.id", 5))
+ .desc(truncate("data", 10))
+ .build();
+ Set<String> sortedCols = SortOrderUtil.orderPreservingSortedColumns(order);
+ Assert.assertEquals(ImmutableSet.of("data"), sortedCols);
+ }
}
diff --git a/data/src/main/java/org/apache/iceberg/data/BaseWriterFactory.java b/data/src/main/java/org/apache/iceberg/data/BaseWriterFactory.java
index 72c0ea5..48b368c 100644
--- a/data/src/main/java/org/apache/iceberg/data/BaseWriterFactory.java
+++ b/data/src/main/java/org/apache/iceberg/data/BaseWriterFactory.java
@@ -85,7 +85,7 @@ public abstract class BaseWriterFactory<T> implements WriterFactory<T> {
OutputFile outputFile = file.encryptingOutputFile();
EncryptionKeyMetadata keyMetadata = file.keyMetadata();
Map<String, String> properties = table.properties();
- MetricsConfig metricsConfig = MetricsConfig.fromProperties(properties);
+ MetricsConfig metricsConfig = MetricsConfig.forTable(table);
try {
switch (dataFileFormat) {
@@ -148,7 +148,7 @@ public abstract class BaseWriterFactory<T> implements WriterFactory<T> {
OutputFile outputFile = file.encryptingOutputFile();
EncryptionKeyMetadata keyMetadata = file.keyMetadata();
Map<String, String> properties = table.properties();
- MetricsConfig metricsConfig = MetricsConfig.fromProperties(properties);
+ MetricsConfig metricsConfig = MetricsConfig.forTable(table);
try {
switch (deleteFileFormat) {
diff --git a/data/src/test/java/org/apache/iceberg/io/TestWriterMetrics.java b/data/src/test/java/org/apache/iceberg/io/TestWriterMetrics.java
new file mode 100644
index 0000000..1cccef3
--- /dev/null
+++ b/data/src/test/java/org/apache/iceberg/io/TestWriterMetrics.java
@@ -0,0 +1,149 @@
+/*
+ * 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.io;
+
+import java.io.File;
+import java.nio.ByteBuffer;
+import java.util.Map;
+import org.apache.iceberg.DataFile;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.PartitionSpec;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.SortOrder;
+import org.apache.iceberg.TableProperties;
+import org.apache.iceberg.TestTables;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Conversions;
+import org.apache.iceberg.types.Types;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+@RunWith(Parameterized.class)
+public abstract class TestWriterMetrics<T> {
+
+ private static final int FORMAT_V2 = 2;
+
+ protected static final Types.NestedField ID_FIELD = required(1, "id", Types.IntegerType.get());
+ protected static final Types.NestedField DATA_FIELD = optional(2, "data", Types.StringType.get());
+
+ protected static final Types.StructType NESTED_FIELDS = Types.StructType.of(
+ required(4, "booleanField", Types.BooleanType.get()),
+ optional(5, "longValue", Types.LongType.get()));
+
+ protected static final Types.NestedField STRUCT_FIELD = optional(3, "structField", NESTED_FIELDS);
+
+ // create a schema with all supported fields
+ protected static final Schema SCHEMA = new Schema(
+ ID_FIELD,
+ DATA_FIELD,
+ STRUCT_FIELD
+ );
+
+ protected static final SortOrder sortOrder =
+ SortOrder.builderFor(SCHEMA).asc("id").asc("structField.longValue").build();
+
+ protected static final Map<String, String> properties =
+ ImmutableMap.of(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none");
+
+ @Rule
+ public TemporaryFolder temp = new TemporaryFolder();
+
+ protected FileFormat fileFormat;
+ protected TestTables.TestTable table = null;
+ protected File metadataDir = null;
+ private OutputFileFactory fileFactory = null;
+
+ @Parameterized.Parameters(name = "FileFormat = {0}")
+ public static Object[][] parameters() {
+ return new Object[][] {
+ {FileFormat.ORC},
+ {FileFormat.PARQUET}
+ };
+ }
+
+ public TestWriterMetrics(FileFormat fileFormat) {
+ this.fileFormat = fileFormat;
+ }
+
+ protected abstract WriterFactory<T> newWriterFactory(Schema dataSchema);
+
+ protected abstract T toRow(Integer id, String data, boolean boolValue, Long longValue);
+
+ @Before
+ public void setupTable() throws Exception {
+ File tableDir = temp.newFolder();
+ tableDir.delete(); // created by table create
+
+ this.metadataDir = new File(tableDir, "metadata");
+
+ this.table = TestTables.create(
+ tableDir,
+ "test",
+ SCHEMA,
+ PartitionSpec.unpartitioned(),
+ sortOrder,
+ FORMAT_V2);
+ table.updateProperties().set(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none").commit();
+
+ this.fileFactory = OutputFileFactory.builderFor(table, 1, 1).format(fileFormat).build();
+ }
+
+ @After
+ public void after() {
+ TestTables.clearTables();
+ }
+
+ @Test
+ public void verifySortedColMetric() throws Exception {
+ T row = toRow(3, "3", true, 3L);
+ DataWriter dataWriter = newWriterFactory(SCHEMA).newDataWriter(
+ fileFactory.newOutputFile(),
+ PartitionSpec.unpartitioned(),
+ null
+ );
+ dataWriter.add(row);
+ dataWriter.close();
+ DataFile dataFile = dataWriter.toDataFile();
+
+ // Only two sorted fields (id, structField.longValue) will have metrics
+ Map<Integer, ByteBuffer> lowerBounds = dataFile.lowerBounds();
+ Assert.assertEquals(3, (int) Conversions.fromByteBuffer(Types.IntegerType.get(), lowerBounds.get(1)));
+ Assert.assertFalse(lowerBounds.containsKey(2));
+ Assert.assertFalse(lowerBounds.containsKey(3));
+ Assert.assertFalse(lowerBounds.containsKey(4));
+ Assert.assertEquals(3L, (long) Conversions.fromByteBuffer(Types.LongType.get(), lowerBounds.get(5)));
+
+ Map<Integer, ByteBuffer> upperBounds = dataFile.upperBounds();
+ Assert.assertEquals(3, (int) Conversions.fromByteBuffer(Types.IntegerType.get(), upperBounds.get(1)));
+ Assert.assertFalse(upperBounds.containsKey(2));
+ Assert.assertFalse(upperBounds.containsKey(3));
+ Assert.assertFalse(upperBounds.containsKey(4));
+ Assert.assertEquals(3L, (long) Conversions.fromByteBuffer(Types.LongType.get(), upperBounds.get(5)));
+ }
+}
diff --git a/data/src/test/java/org/apache/iceberg/orc/TestOrcMetrics.java b/data/src/test/java/org/apache/iceberg/orc/TestOrcMetrics.java
index 2f861a1..975b695 100644
--- a/data/src/test/java/org/apache/iceberg/orc/TestOrcMetrics.java
+++ b/data/src/test/java/org/apache/iceberg/orc/TestOrcMetrics.java
@@ -39,17 +39,27 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableSet;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.iceberg.types.Type;
import org.junit.Assert;
-import org.junit.Rule;
-import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
/**
* Test Metrics for ORC.
*/
+@RunWith(Parameterized.class)
public class TestOrcMetrics extends TestMetrics {
static final ImmutableSet<Object> BINARY_TYPES = ImmutableSet.of(Type.TypeID.BINARY,
Type.TypeID.FIXED, Type.TypeID.UUID);
+ @Parameterized.Parameters(name = "formatVersion = {0}")
+ public static Object[] parameters() {
+ return new Object[] { 1, 2 };
+ }
+
+ public TestOrcMetrics(int formatVersion) {
+ super(formatVersion);
+ }
+
@Override
protected OutputFile createOutputFile() throws IOException {
File tmpFolder = temp.newFolder("orc");
@@ -57,9 +67,6 @@ public class TestOrcMetrics extends TestMetrics {
return Files.localOutput(new File(tmpFolder, FileFormat.ORC.addExtension(filename)));
}
- @Rule
- public TemporaryFolder temp = new TemporaryFolder();
-
@Override
public FileFormat fileFormat() {
return FileFormat.ORC;
diff --git a/data/src/test/java/org/apache/iceberg/parquet/TestParquetMetrics.java b/data/src/test/java/org/apache/iceberg/parquet/TestParquetMetrics.java
index a5c1bff..457d9c3 100644
--- a/data/src/test/java/org/apache/iceberg/parquet/TestParquetMetrics.java
+++ b/data/src/test/java/org/apache/iceberg/parquet/TestParquetMetrics.java
@@ -38,18 +38,25 @@ import org.apache.iceberg.io.OutputFile;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.parquet.hadoop.ParquetFileReader;
-import org.junit.Rule;
-import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
/**
* Test Metrics for Parquet.
*/
+@RunWith(Parameterized.class)
public class TestParquetMetrics extends TestMetrics {
private static final Map<String, String> SMALL_ROW_GROUP_CONFIG = ImmutableMap.of(
TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES, "1600");
- @Rule
- public TemporaryFolder temp = new TemporaryFolder();
+ @Parameterized.Parameters(name = "formatVersion = {0}")
+ public static Object[] parameters() {
+ return new Object[] { 1, 2 };
+ }
+
+ public TestParquetMetrics(int formatVersion) {
+ super(formatVersion);
+ }
@Override
public FileFormat fileFormat() {
diff --git a/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkWriterMetrics.java b/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkWriterMetrics.java
new file mode 100644
index 0000000..df4662e
--- /dev/null
+++ b/flink/src/test/java/org/apache/iceberg/flink/sink/TestFlinkWriterMetrics.java
@@ -0,0 +1,51 @@
+/*
+ * 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.sink;
+
+import org.apache.flink.table.data.GenericRowData;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.data.StringData;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.TestWriterMetrics;
+import org.apache.iceberg.io.WriterFactory;
+
+public class TestFlinkWriterMetrics extends TestWriterMetrics<RowData> {
+
+ public TestFlinkWriterMetrics(FileFormat fileFormat) {
+ super(fileFormat);
+ }
+
+ @Override
+ protected WriterFactory<RowData> newWriterFactory(Schema dataSchema) {
+ return FlinkWriterFactory.builderFor(table)
+ .dataSchema(table.schema())
+ .dataFileFormat(fileFormat)
+ .deleteFileFormat(fileFormat)
+ .build();
+ }
+
+ @Override
+ protected RowData toRow(Integer id, String data, boolean boolValue, Long longValue) {
+ GenericRowData nested = GenericRowData.of(boolValue, longValue);
+ GenericRowData row = GenericRowData.of(id, StringData.fromString(data), nested);
+ return row;
+ }
+}
diff --git a/orc/src/main/java/org/apache/iceberg/orc/ORC.java b/orc/src/main/java/org/apache/iceberg/orc/ORC.java
index 417598b..b326c2f 100644
--- a/orc/src/main/java/org/apache/iceberg/orc/ORC.java
+++ b/orc/src/main/java/org/apache/iceberg/orc/ORC.java
@@ -150,7 +150,7 @@ public class ORC {
schema(table.schema());
withSpec(table.spec());
setAll(table.properties());
- metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
diff --git a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
index aadddbf..1d53d83 100644
--- a/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
+++ b/parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java
@@ -125,7 +125,7 @@ public class Parquet {
public WriteBuilder forTable(Table table) {
schema(table.schema());
setAll(table.properties());
- metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
@@ -391,7 +391,7 @@ public class Parquet {
schema(table.schema());
withSpec(table.spec());
setAll(table.properties());
- metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
@@ -489,7 +489,7 @@ public class Parquet {
rowSchema(table.schema());
withSpec(table.spec());
setAll(table.properties());
- metricsConfig(MetricsConfig.fromProperties(table.properties()));
+ metricsConfig(MetricsConfig.forTable(table));
return this;
}
diff --git a/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java b/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
index 831d7d8..fcfb973 100644
--- a/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
+++ b/spark/src/main/java/org/apache/iceberg/spark/SparkTableUtil.java
@@ -458,7 +458,7 @@ public class SparkTableUtil {
Map<String, String> partition = Collections.emptyMap();
PartitionSpec spec = PartitionSpec.unpartitioned();
Configuration conf = spark.sessionState().newHadoopConf();
- MetricsConfig metricsConfig = MetricsConfig.fromProperties(targetTable.properties());
+ MetricsConfig metricsConfig = MetricsConfig.forTable(targetTable);
String nameMappingString = targetTable.properties().get(TableProperties.DEFAULT_NAME_MAPPING);
NameMapping nameMapping = nameMappingString != null ? NameMappingParser.fromJson(nameMappingString) : null;
diff --git a/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkWriterMetrics.java b/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkWriterMetrics.java
new file mode 100644
index 0000000..6e2dd7c
--- /dev/null
+++ b/spark/src/test/java/org/apache/iceberg/spark/source/TestSparkWriterMetrics.java
@@ -0,0 +1,58 @@
+/*
+ * 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.spark.source;
+
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.TestWriterMetrics;
+import org.apache.iceberg.io.WriterFactory;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.unsafe.types.UTF8String;
+
+public class TestSparkWriterMetrics extends TestWriterMetrics<InternalRow> {
+
+ public TestSparkWriterMetrics(FileFormat fileFormat) {
+ super(fileFormat);
+ }
+
+ @Override
+ protected WriterFactory<InternalRow> newWriterFactory(Schema dataSchema) {
+ return SparkWriterFactory.builderFor(table)
+ .dataSchema(table.schema())
+ .dataFileFormat(fileFormat)
+ .deleteFileFormat(fileFormat)
+ .build();
+ }
+
+ @Override
+ protected InternalRow toRow(Integer id, String data, boolean boolValue, Long longValue) {
+ InternalRow row = new GenericInternalRow(3);
+ row.update(0, id);
+ row.update(1, UTF8String.fromString(data));
+
+ InternalRow nested = new GenericInternalRow(2);
+ nested.update(0, boolValue);
+ nested.update(1, longValue);
+
+ row.update(2, nested);
+ return row;
+ }
+}