Make metrics collection configurable via table properties

core/src/main/java/org/apache/iceberg/MetricsConfig.java

@@ -0,0 +1,61 @@
+/*
+ * 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;
+
+import com.google.common.collect.Maps;
+import java.util.Map;
+import org.apache.iceberg.MetricsModes.MetricsMode;
+
+import static org.apache.iceberg.TableProperties.DEFAULT_WRITE_METRICS_MODE;
+import static org.apache.iceberg.TableProperties.DEFAULT_WRITE_METRICS_MODE_DEFAULT;
+
+public class MetricsConfig {
+
+  private static final String COLUMN_CONF_PREFIX = "write.metadata.metrics.column.";
+
+  private Map<String, MetricsMode> columnModes = Maps.newHashMap();
+  private MetricsMode defaultMode;
+
+  private MetricsConfig() {}
+
+  public static MetricsConfig getDefault() {
+    MetricsConfig spec = new MetricsConfig();
+    spec.defaultMode = MetricsModes.fromString(DEFAULT_WRITE_METRICS_MODE_DEFAULT);
+    return spec;
+  }
+
+  public static MetricsConfig fromProperties(Map<String, String> props) {
+    MetricsConfig spec = new MetricsConfig();
+    props.keySet().stream()
+        .filter(key -> key.startsWith(COLUMN_CONF_PREFIX))
+        .forEach(key -> {
+          MetricsMode mode = MetricsModes.fromString(props.get(key));
+          String columnAlias = key.replaceFirst(COLUMN_CONF_PREFIX, "");
+          spec.columnModes.put(columnAlias, mode);
+        });
+    String defaultModeAsString = props.getOrDefault(DEFAULT_WRITE_METRICS_MODE, DEFAULT_WRITE_METRICS_MODE_DEFAULT);
+    spec.defaultMode = MetricsModes.fromString(defaultModeAsString);
+    return spec;
+  }
+
+  public MetricsMode columnMode(String columnAlias) {
+    return columnModes.getOrDefault(columnAlias, defaultMode);
+  }
+}

core/src/main/java/org/apache/iceberg/MetricsModes.java

@@ -0,0 +1,147 @@
+/*
+ * 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;
+
+import com.google.common.base.Preconditions;
+import java.util.Locale;
+import java.util.Objects;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * This class defines different metrics modes, which allow users to control the collection of
+ * value_counts, null_value_counts, lower_bounds, upper_bounds for different columns in metadata.
+ */
+public class MetricsModes {
+
+  private static final Pattern TRUNCATE = Pattern.compile("truncate\\((\\d+)\\)");
+
+  private MetricsModes() {}
+
+  public static MetricsMode fromString(String mode) {
+    if ("none".equalsIgnoreCase(mode)) {
+      return None.get();
+    } else if ("counts".equalsIgnoreCase(mode)) {
+      return Counts.get();
+    } else if ("full".equalsIgnoreCase(mode)) {
+      return Full.get();
+    }
+
+    Matcher truncateMatcher = TRUNCATE.matcher(mode.toLowerCase(Locale.ENGLISH));
+    if (truncateMatcher.matches()) {
+      int length = Integer.parseInt(truncateMatcher.group(1));
+      return Truncate.withLength(length);
+    }
+
+    throw new IllegalArgumentException("Invalid metrics mode: " + mode);
+  }
+
+  public interface MetricsMode {}
+
+  /**
+   * Under this mode, value_counts, null_value_counts, lower_bounds, upper_bounds are not persisted.
+   */
+  public static class None implements MetricsMode {
+    private static final None INSTANCE = new None();
+
+    public static None get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public String toString() {
+      return "none";
+    }
+  }
+
+  /**
+   * Under this mode, only value_counts, null_value_counts are persisted.
+   */
+  public static class Counts implements MetricsMode {
+    private static final Counts INSTANCE = new Counts();
+
+    public static Counts get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public String toString() {
+      return "counts";
+    }
+  }
+
+  /**
+   * Under this mode, value_counts, null_value_counts and truncated lower_bounds, upper_bounds are persisted.
+   */
+  public static class Truncate implements MetricsMode {
+    private final int length;
+
+    private Truncate(int length) {
+      this.length = length;
+    }
+
+    public static Truncate withLength(int length) {
+      Preconditions.checkArgument(length > 0, "Truncate length should be positive");
+      return new Truncate(length);
+    }
+
+    public int length() {
+      return length;
+    }
+
+    @Override
+    public String toString() {
+      return String.format("truncate(%d)", length);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+      if (this == obj) {
+        return true;
+      }
+      if (obj == null || getClass() != obj.getClass()) {
+        return false;
+      }
+      Truncate truncate = (Truncate) obj;
+      return length == truncate.length;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(length);
+    }
+  }
+
+  /**
+   * Under this mode, value_counts, null_value_counts and full lower_bounds, upper_bounds are persisted.
+   */
+  public static class Full implements MetricsMode {
+    private static final Full INSTANCE = new Full();
+
+    public static Full get() {
+      return INSTANCE;
+    }
+
+    @Override
+    public String toString() {
+      return "full";
+    }
+  }
+}

core/src/main/java/org/apache/iceberg/TableProperties.java

@@ -89,6 +89,6 @@ private TableProperties() {}
   public static final String METADATA_COMPRESSION = "write.metadata.compression-codec";
   public static final String METADATA_COMPRESSION_DEFAULT = "none";
 
-  public static final String WRITE_METADATA_TRUNCATE_BYTES = "write.metadata.truncate-length";
-  public static final int WRITE_METADATA_TRUNCATE_BYTES_DEFAULT = 16;
+  public static final String DEFAULT_WRITE_METRICS_MODE = "write.metadata.metrics.default";
+  public static final String DEFAULT_WRITE_METRICS_MODE_DEFAULT = "truncate(16)";
 }

core/src/test/java/org/apache/iceberg/TestMetricsModes.java

@@ -0,0 +1,54 @@
+/*
+ * 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;
+
+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.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+public class TestMetricsModes {
+
+  @Rule
+  public ExpectedException exceptionRule = ExpectedException.none();
+
+  @Test
+  public void testMetricsModeParsing() {
+    Assert.assertEquals(None.get(), MetricsModes.fromString("none"));
+    Assert.assertEquals(None.get(), MetricsModes.fromString("nOnE"));
+    Assert.assertEquals(Counts.get(), MetricsModes.fromString("counts"));
+    Assert.assertEquals(Counts.get(), MetricsModes.fromString("coUntS"));
+    Assert.assertEquals(Truncate.withLength(1), MetricsModes.fromString("truncate(1)"));
+    Assert.assertEquals(Truncate.withLength(10), MetricsModes.fromString("truNcAte(10)"));
+    Assert.assertEquals(Full.get(), MetricsModes.fromString("full"));
+    Assert.assertEquals(Full.get(), MetricsModes.fromString("FULL"));
+  }
+
+  @Test
+  public void testInvalidTruncationLength() {
+    exceptionRule.expect(IllegalArgumentException.class);
+    exceptionRule.expectMessage("length should be positive");
+    MetricsModes.fromString("truncate(0)");
+  }
+}

parquet/src/main/java/org/apache/iceberg/parquet/Parquet.java

@@ -28,6 +28,7 @@
 import java.util.function.Function;
 import com.google.common.collect.Sets;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.iceberg.MetricsConfig;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.SchemaParser;
 import org.apache.iceberg.Table;
@@ -63,9 +64,6 @@
 import static org.apache.iceberg.TableProperties.PARQUET_PAGE_SIZE_BYTES_DEFAULT;
 import static org.apache.iceberg.TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES;
 import static org.apache.iceberg.TableProperties.PARQUET_ROW_GROUP_SIZE_BYTES_DEFAULT;
-import static org.apache.iceberg.TableProperties.WRITE_METADATA_TRUNCATE_BYTES;
-import static org.apache.iceberg.TableProperties.WRITE_METADATA_TRUNCATE_BYTES_DEFAULT;
-
 
 public class Parquet {
   private Parquet() {
@@ -86,6 +84,7 @@ public static class WriteBuilder {
     private Map<String, String> metadata = Maps.newLinkedHashMap();
     private Map<String, String> config = Maps.newLinkedHashMap();
     private Function<MessageType, ParquetValueWriter<?>> createWriterFunc = null;
+    private MetricsConfig metricsConfig = MetricsConfig.getDefault();
 
     private WriteBuilder(OutputFile file) {
       this.file = file;
@@ -94,6 +93,7 @@ private WriteBuilder(OutputFile file) {
     public WriteBuilder forTable(Table table) {
       schema(table.schema());
       setAll(table.properties());
+      metricsConfig(MetricsConfig.fromProperties(table.properties()));
       return this;
     }
 
@@ -133,6 +133,11 @@ public WriteBuilder createWriterFunc(
       return this;
     }
 
+    public WriteBuilder metricsConfig(MetricsConfig newMetricsConfig) {
+      this.metricsConfig = newMetricsConfig;
+      return this;
+    }
+
     @SuppressWarnings("unchecked")
     private <T> WriteSupport<T> getWriteSupport(MessageType type) {
       if (writeSupport != null) {
@@ -168,9 +173,6 @@ public <D> FileAppender<D> build() throws IOException {
           PARQUET_PAGE_SIZE_BYTES, PARQUET_PAGE_SIZE_BYTES_DEFAULT));
       int dictionaryPageSize = Integer.parseInt(config.getOrDefault(
           PARQUET_DICT_SIZE_BYTES, PARQUET_DICT_SIZE_BYTES_DEFAULT));
-      int statsTruncateLength = Integer.parseInt(config.getOrDefault(
-          WRITE_METADATA_TRUNCATE_BYTES, String.valueOf(WRITE_METADATA_TRUNCATE_BYTES_DEFAULT)));
-
 
       WriterVersion writerVersion = WriterVersion.PARQUET_1_0;
 
@@ -198,8 +200,8 @@ public <D> FileAppender<D> build() throws IOException {
             .build();
 
         return new org.apache.iceberg.parquet.ParquetWriter<>(
-            conf, file, schema, rowGroupSize, statsTruncateLength, metadata,
-                createWriterFunc, codec(), parquetProperties);
+            conf, file, schema, rowGroupSize, metadata, createWriterFunc, codec(),
+            parquetProperties, metricsConfig);
       } else {
         return new ParquetWriteAdapter<>(new ParquetWriteBuilder<D>(ParquetIO.file(file))
             .withWriterVersion(writerVersion)
@@ -212,7 +214,8 @@ public <D> FileAppender<D> build() throws IOException {
             .withRowGroupSize(rowGroupSize)
             .withPageSize(pageSize)
             .withDictionaryPageSize(dictionaryPageSize)
-            .build(), statsTruncateLength);
+            .build(),
+            metricsConfig);
       }
     }
   }