Skip to content

Commit 7a1c5b1

Browse files
mao-liuthe-other-tim-brown
mao-liu
and
the-other-tim-brown
authored
[755] Support column stats for paimon (#767)
* paimon: wip column stats * paimon column stats: fix tests * paimon: fixing some tests * paimon: fixing more tests * paimon: fix column stats column checks * paimon: stats: handling timestamp as long * paimon: add to xtable-utilities * paimon: lint * paimon: ticked version to 1.3.1 * paimon: reducing logging * paimon: annotating assumptions with comments * paimon: expanded tests for stats extractor * paimon: stats extractor to its own class * paimon: handling deleted stats * paimon: spotless * paimon: fix tests * paimon: tests remove wildcard imports * paimon: assumptions validated * paimon: code review changes * paimon: added test case for lack of stats for complex fields * paimon: extended tests and updated test TODO with github issue reference * paimon: fix tests * pin catalyst version * paimon: fix tests in xtable-utilities --------- Co-authored-by: Timothy Brown <tim@onehouse.ai>
1 parent 50ae611 commit 7a1c5b1

9 files changed

Lines changed: 836 additions & 83 deletions

File tree

‎pom.xml‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -89,7 +89,7 @@
8989
<spark.version.prefix>3.4</spark.version.prefix>
9090
<iceberg.version>1.4.2</iceberg.version>
9191
<delta.version>2.4.0</delta.version>
92-
<paimon.version>1.2.0</paimon.version>
92+
<paimon.version>1.3.1</paimon.version>
9393
<jackson.version>2.18.2</jackson.version>
9494
<spotless.version>2.43.0</spotless.version>
9595
<apache.rat.version>0.16.1</apache.rat.version>
@@ -374,6 +374,12 @@
374374
<version>${spark.version}</version>
375375
<scope>provided</scope>
376376
</dependency>
377+
<dependency>
378+
<groupId>org.apache.spark</groupId>
379+
<artifactId>spark-catalyst_${scala.binary.version}</artifactId>
380+
<version>${spark.version}</version>
381+
<scope>provided</scope>
382+
</dependency>
377383

378384
<dependency>
379385
<groupId>commons-cli</groupId>

‎xtable-core/src/main/java/org/apache/xtable/paimon/PaimonDataFileExtractor.java‎

Lines changed: 3 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919
package org.apache.xtable.paimon;
2020

2121
import java.util.ArrayList;
22-
import java.util.Collections;
2322
import java.util.HashSet;
2423
import java.util.Iterator;
2524
import java.util.List;
@@ -29,7 +28,6 @@
2928
import lombok.extern.log4j.Log4j2;
3029

3130
import org.apache.paimon.Snapshot;
32-
import org.apache.paimon.io.DataFileMeta;
3331
import org.apache.paimon.manifest.FileKind;
3432
import org.apache.paimon.manifest.ManifestEntry;
3533
import org.apache.paimon.manifest.ManifestFile;
@@ -39,7 +37,6 @@
3937
import org.apache.paimon.table.source.snapshot.SnapshotReader;
4038

4139
import org.apache.xtable.model.schema.InternalSchema;
42-
import org.apache.xtable.model.stat.ColumnStat;
4340
import org.apache.xtable.model.storage.InternalDataFile;
4441
import org.apache.xtable.model.storage.InternalFilesDiff;
4542

@@ -49,6 +46,8 @@ public class PaimonDataFileExtractor {
4946
private final PaimonPartitionExtractor partitionExtractor =
5047
PaimonPartitionExtractor.getInstance();
5148

49+
private final PaimonStatsExtractor statsExtractor = PaimonStatsExtractor.getInstance();
50+
5251
private static final PaimonDataFileExtractor INSTANCE = new PaimonDataFileExtractor();
5352

5453
public static PaimonDataFileExtractor getInstance() {
@@ -84,7 +83,7 @@ public InternalDataFile toInternalDataFile(
8483
.recordCount(entry.file().rowCount())
8584
.partitionValues(
8685
partitionExtractor.toPartitionValues(table, entry.partition(), internalSchema))
87-
.columnStats(toColumnStats(entry.file()))
86+
.columnStats(statsExtractor.extractColumnStats(entry.file(), internalSchema))
8887
.build();
8988
}
9089

@@ -101,12 +100,6 @@ private String toFullPhysicalPath(FileStoreTable table, ManifestEntry entry) {
101100
}
102101
}
103102

104-
private List<ColumnStat> toColumnStats(DataFileMeta file) {
105-
// TODO: Implement logic to extract column stats from the file meta
106-
// https://github.com/apache/incubator-xtable/issues/755
107-
return Collections.emptyList();
108-
}
109-
110103
/**
111104
* Extracts file changes (added and removed files) from delta manifests for a given snapshot. This
112105
* method reads only the delta manifests which contain the changes introduced in this specific

‎xtable-core/src/main/java/org/apache/xtable/paimon/PaimonStatsExtractor.java‎

Lines changed: 187 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,187 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
19+
package org.apache.xtable.paimon;
20+
21+
import java.util.ArrayList;
22+
import java.util.List;
23+
import java.util.Map;
24+
import java.util.stream.Collectors;
25+
26+
import lombok.AccessLevel;
27+
import lombok.NoArgsConstructor;
28+
import lombok.extern.log4j.Log4j2;
29+
30+
import org.apache.paimon.data.BinaryArray;
31+
import org.apache.paimon.data.BinaryRow;
32+
import org.apache.paimon.data.Timestamp;
33+
import org.apache.paimon.io.DataFileMeta;
34+
import org.apache.paimon.stats.SimpleStats;
35+
import org.apache.paimon.types.TimestampType;
36+
37+
import org.apache.xtable.exception.ReadException;
38+
import org.apache.xtable.model.schema.InternalField;
39+
import org.apache.xtable.model.schema.InternalSchema;
40+
import org.apache.xtable.model.schema.InternalType;
41+
import org.apache.xtable.model.stat.ColumnStat;
42+
import org.apache.xtable.model.stat.Range;
43+
44+
@Log4j2
45+
@NoArgsConstructor(access = AccessLevel.PRIVATE)
46+
public class PaimonStatsExtractor {
47+
private static final PaimonStatsExtractor INSTANCE = new PaimonStatsExtractor();
48+
49+
public static PaimonStatsExtractor getInstance() {
50+
return INSTANCE;
51+
}
52+
53+
public List<ColumnStat> extractColumnStats(DataFileMeta file, InternalSchema internalSchema) {
54+
List<ColumnStat> columnStats = new ArrayList<>();
55+
Map<String, InternalField> fieldMap =
56+
internalSchema.getAllFields().stream()
57+
.collect(Collectors.toMap(InternalField::getPath, f -> f));
58+
59+
// stats for all columns are present in valueStats, we can safely ignore file.keyStats()
60+
SimpleStats valueStats = file.valueStats();
61+
if (valueStats != null) {
62+
List<String> colNames = file.valueStatsCols();
63+
if (colNames == null) {
64+
// if column names are not present, then stats are being collected for all columns
65+
colNames =
66+
internalSchema.getAllFields().stream()
67+
.map(InternalField::getPath)
68+
.collect(Collectors.toList());
69+
}
70+
71+
if (colNames.size() != valueStats.minValues().getFieldCount()) {
72+
// paranoia check - this should never happen, but if the code reaches here, then there is a
73+
// bug! Please file a bug report
74+
throw new ReadException(
75+
String.format(
76+
"Mismatch between column stats names and values arity: names=%d, values=%d",
77+
colNames.size(), valueStats.minValues().getFieldCount()));
78+
}
79+
80+
collectColumnStats(columnStats, valueStats, colNames, fieldMap, file.rowCount());
81+
}
82+
83+
return columnStats;
84+
}
85+
86+
private void collectColumnStats(
87+
List<ColumnStat> columnStats,
88+
SimpleStats stats,
89+
List<String> colNames,
90+
Map<String, InternalField> fieldMap,
91+
long rowCount) {
92+
93+
BinaryRow minValues = stats.minValues();
94+
BinaryRow maxValues = stats.maxValues();
95+
BinaryArray nullCounts = stats.nullCounts();
96+
97+
for (int i = 0; i < colNames.size(); i++) {
98+
String colName = colNames.get(i);
99+
InternalField field = fieldMap.get(colName);
100+
if (field == null) {
101+
continue;
102+
}
103+
104+
// Check if we already have stats for this field
105+
boolean alreadyExists =
106+
columnStats.stream().anyMatch(cs -> cs.getField().getPath().equals(colName));
107+
if (alreadyExists) {
108+
continue;
109+
}
110+
111+
InternalType type = field.getSchema().getDataType();
112+
Object min = getValue(minValues, i, type, field.getSchema());
113+
Object max = getValue(maxValues, i, type, field.getSchema());
114+
long nullCount = (nullCounts != null && i < nullCounts.size()) ? nullCounts.getLong(i) : 0L;
115+
116+
columnStats.add(
117+
ColumnStat.builder()
118+
.field(field)
119+
.range(min != null && max != null ? Range.vector(min, max) : null)
120+
.numNulls(nullCount)
121+
.numValues(rowCount)
122+
.build());
123+
}
124+
}
125+
126+
private Object getValue(BinaryRow row, int index, InternalType type, InternalSchema fieldSchema) {
127+
if (row.isNullAt(index)) {
128+
return null;
129+
}
130+
switch (type) {
131+
case BOOLEAN:
132+
return row.getBoolean(index);
133+
case INT:
134+
case DATE:
135+
return row.getInt(index);
136+
case LONG:
137+
return row.getLong(index);
138+
case TIMESTAMP:
139+
case TIMESTAMP_NTZ:
140+
int tsPrecision;
141+
InternalSchema.MetadataValue tsPrecisionEnum =
142+
(InternalSchema.MetadataValue)
143+
fieldSchema.getMetadata().get(InternalSchema.MetadataKey.TIMESTAMP_PRECISION);
144+
if (tsPrecisionEnum == InternalSchema.MetadataValue.MILLIS) {
145+
tsPrecision = 3;
146+
} else if (tsPrecisionEnum == InternalSchema.MetadataValue.MICROS) {
147+
tsPrecision = 6;
148+
} else if (tsPrecisionEnum == InternalSchema.MetadataValue.NANOS) {
149+
tsPrecision = 9;
150+
} else {
151+
log.warn(
152+
"Field idx={}, name={} does not have MetadataKey.TIMESTAMP_PRECISION set, defaulting to default precision",
153+
index,
154+
fieldSchema.getName());
155+
tsPrecision = TimestampType.DEFAULT_PRECISION;
156+
}
157+
Timestamp ts = row.getTimestamp(index, tsPrecision);
158+
159+
// according to docs for org.apache.xtable.model.stat.Range, timestamp is stored as millis
160+
// or micros - even if precision is higher than micros, return micros
161+
if (tsPrecisionEnum == InternalSchema.MetadataValue.MILLIS) {
162+
return ts.getMillisecond();
163+
} else {
164+
return ts.toMicros();
165+
}
166+
case FLOAT:
167+
return row.getFloat(index);
168+
case DOUBLE:
169+
return row.getDouble(index);
170+
case STRING:
171+
case ENUM:
172+
return row.getString(index).toString();
173+
case DECIMAL:
174+
int precision =
175+
(int) fieldSchema.getMetadata().get(InternalSchema.MetadataKey.DECIMAL_PRECISION);
176+
int scale = (int) fieldSchema.getMetadata().get(InternalSchema.MetadataKey.DECIMAL_SCALE);
177+
return row.getDecimal(index, precision, scale).toBigDecimal();
178+
default:
179+
log.warn(
180+
"Handling of {}-type stats for column idx={}, name={} is not yet implemented, skipping stats for this column",
181+
type,
182+
index,
183+
fieldSchema.getName());
184+
return null;
185+
}
186+
}
187+
}

‎xtable-core/src/test/java/org/apache/xtable/TestPaimonTable.java‎

Lines changed: 42 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -70,9 +70,21 @@ public static GenericTable<GenericRow, String> createTable(
7070
Path tempDir,
7171
Configuration hadoopConf,
7272
boolean additionalColumns) {
73+
74+
Schema schema = buildGenericSchema(partitionField, additionalColumns);
75+
return createTable(tableName, partitionField, tempDir, hadoopConf, additionalColumns, schema);
76+
}
77+
78+
public static GenericTable<GenericRow, String> createTable(
79+
String tableName,
80+
String partitionField,
81+
Path tempDir,
82+
Configuration hadoopConf,
83+
boolean additionalColumns,
84+
Schema schema) {
7385
String basePath = initBasePath(tempDir, tableName);
7486
Catalog catalog = createFilesystemCatalog(basePath, hadoopConf);
75-
FileStoreTable paimonTable = createTable(catalog, partitionField, additionalColumns);
87+
FileStoreTable paimonTable = createTable(catalog, tableName, schema);
7688

7789
System.out.println(
7890
"Initialized Paimon test table at base path: "
@@ -90,20 +102,18 @@ public static Catalog createFilesystemCatalog(String basePath, Configuration had
90102
return CatalogFactory.createCatalog(context);
91103
}
92104

93-
public static FileStoreTable createTable(
94-
Catalog catalog, String partitionField, boolean additionalColumns) {
105+
public static FileStoreTable createTable(Catalog catalog, String tableName, Schema schema) {
95106
try {
96107
catalog.createDatabase("test_db", true);
97-
Identifier identifier = Identifier.create("test_db", "test_table");
98-
Schema schema = buildSchema(partitionField, additionalColumns);
108+
Identifier identifier = Identifier.create("test_db", tableName);
99109
catalog.createTable(identifier, schema, true);
100110
return (FileStoreTable) catalog.getTable(identifier);
101111
} catch (Exception e) {
102112
throw new RuntimeException(e);
103113
}
104114
}
105115

106-
private static Schema buildSchema(String partitionField, boolean additionalColumns) {
116+
private static Schema buildGenericSchema(String partitionField, boolean additionalColumns) {
107117
Schema.Builder builder =
108118
Schema.newBuilder()
109119
.primaryKey("id")
@@ -116,7 +126,8 @@ private static Schema buildSchema(String partitionField, boolean additionalColum
116126
.column("description", DataTypes.VARCHAR(255))
117127
.option("bucket", "1")
118128
.option("bucket-key", "id")
119-
.option("full-compaction.delta-commits", "1");
129+
.option("full-compaction.delta-commits", "1")
130+
.option("metadata.stats-mode", "full");
120131

121132
if (partitionField != null) {
122133
builder
@@ -178,20 +189,12 @@ public List<GenericRow> insertRecordsForSpecialPartition(int numRows) {
178189
}
179190

180191
private List<GenericRow> insertRecordsToPartition(int numRows, String partitionValue) {
181-
BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
182-
try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
183-
List<GenericRow> rows = new ArrayList<>(numRows);
184-
for (int i = 0; i < numRows; i++) {
185-
GenericRow row = buildGenericRow(i, paimonTable.schema(), partitionValue);
186-
writer.write(row);
187-
rows.add(row);
188-
}
189-
commitWrites(batchWriteBuilder, writer);
190-
compactTable();
191-
return rows;
192-
} catch (Exception e) {
193-
throw new RuntimeException("Failed to insert rows into Paimon table", e);
192+
List<GenericRow> rows = new ArrayList<>(numRows);
193+
for (int i = 0; i < numRows; i++) {
194+
rows.add(buildGenericRow(i, paimonTable.schema(), partitionValue));
194195
}
196+
writeRows(paimonTable, rows);
197+
return rows;
195198
}
196199

197200
@Override
@@ -224,8 +227,12 @@ public void deleteRows(List<GenericRow> rows) {
224227
}
225228

226229
private void compactTable() {
227-
BatchWriteBuilder batchWriteBuilder = paimonTable.newBatchWriteBuilder();
228-
SnapshotReader snapshotReader = paimonTable.newSnapshotReader();
230+
compactTable(paimonTable);
231+
}
232+
233+
public static void compactTable(FileStoreTable table) {
234+
BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
235+
SnapshotReader snapshotReader = table.newSnapshotReader();
229236
try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
230237
for (BucketEntry bucketEntry : snapshotReader.bucketEntries()) {
231238
writer.compact(bucketEntry.partition(), bucketEntry.bucket(), true);
@@ -236,6 +243,19 @@ private void compactTable() {
236243
}
237244
}
238245

246+
public static void writeRows(FileStoreTable table, List<GenericRow> rows) {
247+
BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
248+
try (BatchTableWrite writer = batchWriteBuilder.newWrite()) {
249+
for (GenericRow row : rows) {
250+
writer.write(row);
251+
}
252+
commitWrites(batchWriteBuilder, writer);
253+
compactTable(table);
254+
} catch (Exception e) {
255+
throw new RuntimeException("Failed to write rows into Paimon table", e);
256+
}
257+
}
258+
239259
private static void commitWrites(BatchWriteBuilder batchWriteBuilder, BatchTableWrite writer)
240260
throws Exception {
241261
BatchTableCommit commit = batchWriteBuilder.newCommit();

0 commit comments

Comments
 (0)