This is an automated email from the ASF dual-hosted git repository.
rdblue pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new e050d4d798 Core: Make MetricsConfig.forTable collect full metrics for
order preserving transform source fields (#17960)
e050d4d798 is described below
commit e050d4d798d2241c40a6caf795d5098cea24b6d5
Author: Amogh Jahagirdar <[email protected]>
AuthorDate: Tue Sep 29 17:59:57 2026 -0600
Core: Make MetricsConfig.forTable collect full metrics for order preserving
transform source fields (#17960)
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../java/org/apache/iceberg/MetricsConfig.java | 28 +++++++-
.../java/org/apache/iceberg/MetricsTestUtil.java | 6 ++
.../java/org/apache/iceberg/TestMetricsConfig.java | 83 ++++++++++++++++++++++
.../spark/source/TestWriteMetricsConfig.java | 2 +-
.../spark/source/TestWriteMetricsConfig.java | 2 +-
.../spark/source/TestWriteMetricsConfig.java | 2 +-
.../spark/source/TestWriteMetricsConfig.java | 2 +-
7 files changed, 120 insertions(+), 5 deletions(-)
diff --git a/core/src/main/java/org/apache/iceberg/MetricsConfig.java
b/core/src/main/java/org/apache/iceberg/MetricsConfig.java
index 273d51f888..eba0eb7cee 100644
--- a/core/src/main/java/org/apache/iceberg/MetricsConfig.java
+++ b/core/src/main/java/org/apache/iceberg/MetricsConfig.java
@@ -118,7 +118,7 @@ public final class MetricsConfig implements Serializable {
* @return a metrics config for the given table
*/
public static MetricsConfig forTable(Table table) {
- return from(table.properties(), table.schema(), table.sortOrder());
+ return from(table.properties(), table.schema(), table.sortOrder(),
table.spec());
}
public Iterable<Integer> metricsFieldIds() {
@@ -257,6 +257,11 @@ public final class MetricsConfig implements Serializable {
*/
@Deprecated
public static MetricsConfig from(Map<String, String> props, Schema schema,
SortOrder order) {
+ return from(props, schema, order, null);
+ }
+
+ private static MetricsConfig from(
+ Map<String, String> props, Schema schema, SortOrder order, PartitionSpec
spec) {
int maxDefaultColumns = maxInferredColumnDefaults(props);
// Handle configured default mode
@@ -286,6 +291,9 @@ public final class MetricsConfig implements Serializable {
// Override automatic modes with configured modes
columnModes.putAll(configuredColumnModes(props));
+ // Force full metrics for partition source columns
+ columnModes.putAll(partitionColumnModes(spec));
+
Map<Integer, String> idToName = idToName(schema, columnModes);
return new MetricsConfig(columnModes, defaultMode, idToName);
@@ -338,6 +346,24 @@ public final class MetricsConfig implements Serializable {
return builder.build();
}
+ private static Map<String, MetricsMode> partitionColumnModes(PartitionSpec
spec) {
+ if (spec == null) {
+ return ImmutableMap.of();
+ }
+
+ ImmutableMap.Builder<String, MetricsMode> builder = ImmutableMap.builder();
+ for (PartitionField field : spec.fields()) {
+ if (field.transform().preservesOrder()) {
+ String name = spec.schema().findColumnName(field.sourceId());
+ // truncate[W] could use stats truncated to at least W, but full is
used for simplicity
+ builder.put(name, MetricsModes.Full.get());
+ }
+ }
+
+ // multiple partition fields can share a source column, so allow duplicate
keys
+ return builder.buildKeepingLast();
+ }
+
private static Map<Integer, String> idToName(
Schema schema, Map<String, MetricsMode> columnModes) {
if (schema != null) {
diff --git a/core/src/test/java/org/apache/iceberg/MetricsTestUtil.java
b/core/src/test/java/org/apache/iceberg/MetricsTestUtil.java
index db0874d111..d752884c6a 100644
--- a/core/src/test/java/org/apache/iceberg/MetricsTestUtil.java
+++ b/core/src/test/java/org/apache/iceberg/MetricsTestUtil.java
@@ -30,10 +30,16 @@ public class MetricsTestUtil {
public static MetricsConfig from(
Map<String, String> properties, Schema schema, SortOrder sortOrder) {
+ return from(properties, schema, sortOrder, null);
+ }
+
+ public static MetricsConfig from(
+ Map<String, String> properties, Schema schema, SortOrder sortOrder,
PartitionSpec spec) {
Table mock = Mockito.mock(Table.class);
Mockito.when(mock.properties()).thenReturn(properties);
Mockito.when(mock.schema()).thenReturn(schema);
Mockito.when(mock.sortOrder()).thenReturn(sortOrder);
+ Mockito.when(mock.spec()).thenReturn(spec);
return MetricsConfig.forTable(mock);
}
}
diff --git a/core/src/test/java/org/apache/iceberg/TestMetricsConfig.java
b/core/src/test/java/org/apache/iceberg/TestMetricsConfig.java
index cb4459472c..7f85326f0c 100644
--- a/core/src/test/java/org/apache/iceberg/TestMetricsConfig.java
+++ b/core/src/test/java/org/apache/iceberg/TestMetricsConfig.java
@@ -30,6 +30,89 @@ import org.junit.jupiter.api.Test;
public class TestMetricsConfig {
+ private static final int ID = 1;
+ private static final int EVENT_TIME = 2;
+ private static final int CATEGORY = 3;
+ private static final int DATA = 4;
+
+ private static final Schema SCHEMA =
+ new Schema(
+ required(ID, "id", Types.IntegerType.get()),
+ optional(EVENT_TIME, "event_time",
Types.TimestampType.withoutZone()),
+ optional(CATEGORY, "category", Types.StringType.get()),
+ optional(DATA, "data", Types.StringType.get()));
+
+ @Test
+ void configCannotDisablePartitionSourceMetrics() {
+ PartitionSpec spec =
PartitionSpec.builderFor(SCHEMA).identity("category").build();
+ Map<String, String> props =
+ ImmutableMap.of(TableProperties.METRICS_MODE_COLUMN_CONF_PREFIX +
"category", "none");
+ MetricsConfig config = MetricsTestUtil.from(props, SCHEMA,
SortOrder.unsorted(), spec);
+
+ assertThat(config.columnMode(CATEGORY))
+ .as("column config should not be able to disable metrics for a
partition source column")
+ .isEqualTo(MetricsModes.Full.get());
+ }
+
+ @Test
+ void bucketPartitionColumnIgnored() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).bucket("category",
4).build();
+ MetricsConfig config =
+ MetricsTestUtil.from(ImmutableMap.of(), SCHEMA, SortOrder.unsorted(),
spec);
+
+ assertThat(config.columnMode(CATEGORY))
+ .as("non-order-preserving partition transform should not promote its
source column")
+ .isEqualTo(MetricsModes.Truncate.withLength(16));
+ }
+
+ @Test
+ void truncatePartitionColumnFullMetrics() {
+ PartitionSpec spec = PartitionSpec.builderFor(SCHEMA).truncate("category",
4).build();
+ MetricsConfig config =
+ MetricsTestUtil.from(ImmutableMap.of(), SCHEMA, SortOrder.unsorted(),
spec);
+
+ assertThat(config.columnMode(CATEGORY))
+ .as("truncate partition source column should get full metrics")
+ .isEqualTo(MetricsModes.Full.get());
+ }
+
+ @Test
+ void multiplePartitionFieldsForSameSourceColumn() {
+ PartitionSpec spec =
+
PartitionSpec.builderFor(SCHEMA).identity("category").truncate("category",
4).build();
+ MetricsConfig config =
+ MetricsTestUtil.from(ImmutableMap.of(), SCHEMA, SortOrder.unsorted(),
spec);
+
+ assertThat(config.columnMode(CATEGORY))
+ .as("source column of multiple partition fields should get full
metrics")
+ .isEqualTo(MetricsModes.Full.get());
+ }
+
+ @Test
+ void columnModeAndFieldIdsFromPartitionSpec() {
+ PartitionSpec spec =
+
PartitionSpec.builderFor(SCHEMA).day("event_time").identity("category").build();
+ MetricsConfig config =
+ MetricsTestUtil.from(ImmutableMap.of(), SCHEMA, SortOrder.unsorted(),
spec);
+
+ assertThat(config.metricsFieldIds())
+ .as("Should track field ids for partition and non-partition columns")
+ .containsExactlyInAnyOrder(ID, EVENT_TIME, CATEGORY, DATA);
+
+ assertThat(config.columnMode(EVENT_TIME))
+ .as("day partition source column should get full metrics")
+ .isEqualTo(MetricsModes.Full.get());
+ assertThat(config.columnMode(CATEGORY))
+ .as("identity partition source column should get full metrics")
+ .isEqualTo(MetricsModes.Full.get());
+ assertThat(config.columnMode(ID))
+ .as("non-partition column should keep the default mode")
+ .isEqualTo(MetricsModes.Truncate.withLength(16));
+ assertThat(config.columnMode(DATA))
+ .as("non-partition column should keep the default mode")
+ .isEqualTo(MetricsModes.Truncate.withLength(16));
+ }
+
@Test
public void testNestedStruct() {
Schema schema =
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
index 45ff918456..951c69ceb3 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
@@ -246,7 +246,7 @@ public class TestWriteMetricsConfig {
String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
- PartitionSpec spec =
PartitionSpec.builderFor(COMPLEX_SCHEMA).identity("strCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
Map<String, String> properties = Maps.newHashMap();
properties.put(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none");
properties.put("write.metadata.metrics.column.longCol", "counts");
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
index ab2479d610..3f4e4a2162 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
@@ -247,7 +247,7 @@ public class TestWriteMetricsConfig {
String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
- PartitionSpec spec =
PartitionSpec.builderFor(COMPLEX_SCHEMA).identity("strCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
Map<String, String> properties = Maps.newHashMap();
properties.put(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none");
properties.put("write.metadata.metrics.column.longCol", "counts");
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
index ab2479d610..3f4e4a2162 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
@@ -247,7 +247,7 @@ public class TestWriteMetricsConfig {
String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
- PartitionSpec spec =
PartitionSpec.builderFor(COMPLEX_SCHEMA).identity("strCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
Map<String, String> properties = Maps.newHashMap();
properties.put(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none");
properties.put("write.metadata.metrics.column.longCol", "counts");
diff --git
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
index ab2479d610..3f4e4a2162 100644
---
a/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
+++
b/spark/v4.2/spark/src/test/java/org/apache/iceberg/spark/source/TestWriteMetricsConfig.java
@@ -247,7 +247,7 @@ public class TestWriteMetricsConfig {
String tableLocation = temp.resolve("iceberg-table").toFile().toString();
HadoopTables tables = new HadoopTables(CONF);
- PartitionSpec spec =
PartitionSpec.builderFor(COMPLEX_SCHEMA).identity("strCol").build();
+ PartitionSpec spec = PartitionSpec.unpartitioned();
Map<String, String> properties = Maps.newHashMap();
properties.put(TableProperties.DEFAULT_WRITE_METRICS_MODE, "none");
properties.put("write.metadata.metrics.column.longCol", "counts");
