You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by lz...@apache.org on 2024/01/23 02:44:54 UTC
(incubator-paimon) branch master updated: [spark] Support ANALYZE table (#2752)
This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 5749a73ed [spark] Support ANALYZE table (#2752)
5749a73ed is described below
commit 5749a73ed79e78336d78528e8568431ecc118847
Author: Zouxxyy <zo...@alibaba-inc.com>
AuthorDate: Tue Jan 23 10:44:51 2024 +0800
[spark] Support ANALYZE table (#2752)
---
.../apache/paimon/operation/FileStoreCommit.java | 4 +-
.../paimon/operation/FileStoreCommitImpl.java | 6 +-
.../java/org/apache/paimon/stats/ColStats.java | 29 +--
.../paimon/stats/{Stats.java => Statistics.java} | 19 +-
.../java/org/apache/paimon/stats/StatsFile.java | 6 +-
.../org/apache/paimon/stats/StatsFileHandler.java | 10 +-
.../paimon/table/AbstractFileStoreTable.java | 11 +
.../org/apache/paimon/table/ReadonlyTable.java | 6 +
.../main/java/org/apache/paimon/table/Table.java | 5 +
.../paimon/operation/FileStoreCommitTest.java | 16 +-
.../org/apache/paimon/stats/StatsFileTest.java | 2 +-
.../apache/paimon/spark/sql/AnalyzeTableTest.scala | 20 ++
.../apache/paimon/spark/sql/AnalyzeTableTest.scala | 20 ++
.../apache/paimon/spark/sql/AnalyzeTableTest.scala | 20 ++
.../spark/catalyst/analysis/PaimonAnalysis.scala | 33 ++-
.../commands/PaimonAnalyzeTableColumnCommand.scala | 165 ++++++++++++++
.../main/scala/org/apache/spark/sql/Utils.scala | 47 +++-
.../paimon/spark/sql/AnalyzeTableTestBase.scala | 247 +++++++++++++++++++++
18 files changed, 617 insertions(+), 49 deletions(-)
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index ac13663f7..fad4c2b70 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -22,7 +22,7 @@ import org.apache.paimon.Snapshot;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.manifest.ManifestCommittable;
import org.apache.paimon.operation.metrics.CommitMetrics;
-import org.apache.paimon.stats.Stats;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.utils.FileStorePathFactory;
@@ -90,7 +90,7 @@ public interface FileStoreCommit {
* Commit new statistics. The {@link Snapshot.CommitKind} of generated snapshot is {@link
* Snapshot.CommitKind#ANALYZE}.
*/
- void commitStatistics(Stats stats, long commitIdentifier);
+ void commitStatistics(Statistics stats, long commitIdentifier);
FileStorePathFactory pathFactory();
diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 37724b35c..1962edcff 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -40,7 +40,7 @@ import org.apache.paimon.options.MemorySize;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.SchemaManager;
-import org.apache.paimon.stats.Stats;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.table.sink.CommitMessage;
import org.apache.paimon.table.sink.CommitMessageImpl;
@@ -513,7 +513,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
}
@Override
- public void commitStatistics(Stats stats, long commitIdentifier) {
+ public void commitStatistics(Statistics stats, long commitIdentifier) {
String statsFileName = statsFileHandler.writeStats(stats);
tryCommit(
Collections.emptyList(),
@@ -786,7 +786,7 @@ public class FileStoreCommitImpl implements FileStoreCommit {
if (newStatsFileName != null) {
statsFileName = newStatsFileName;
} else if (latestSnapshot != null) {
- Optional<Stats> previousStatistic = statsFileHandler.readStats(latestSnapshot);
+ Optional<Statistics> previousStatistic = statsFileHandler.readStats(latestSnapshot);
if (previousStatistic.isPresent()) {
if (previousStatistic.get().schemaId() != latestSchemaId) {
LOG.warn("Schema changed, stats will not be inherited");
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java b/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
index 81a7e0a38..35baf1a5b 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/ColStats.java
@@ -18,6 +18,7 @@
package org.apache.paimon.stats;
+import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.data.serializer.Serializer;
import org.apache.paimon.types.DataType;
@@ -48,6 +49,7 @@ import java.util.OptionalLong;
*
* @param <T> col internal data type
*/
+@Experimental
public class ColStats<T> {
private static final String FIELD_COL_ID = "colId";
@@ -89,6 +91,7 @@ public class ColStats<T> {
@JsonProperty(FIELD_MAX_LEN)
private final @Nullable Long maxLen;
+ // This should only be used by jackson
@JsonCreator
public ColStats(
@JsonProperty(FIELD_COL_ID) int colId,
@@ -107,7 +110,7 @@ public class ColStats<T> {
this.maxLen = maxLen;
}
- public ColStats(
+ private ColStats(
int colId,
@Nullable Long distinctCount,
@Nullable Comparable<T> min,
@@ -124,6 +127,17 @@ public class ColStats<T> {
this.maxLen = maxLen;
}
+ public static <T> ColStats<T> newColStats(
+ int colId,
+ @Nullable Long distinctCount,
+ @Nullable Comparable<T> min,
+ @Nullable Comparable<T> max,
+ @Nullable Long nullCount,
+ @Nullable Long avgLen,
+ @Nullable Long maxLen) {
+ return new ColStats<>(colId, distinctCount, min, max, nullCount, avgLen, maxLen);
+ }
+
public int colId() {
return colId;
}
@@ -189,9 +203,7 @@ public class ColStats<T> {
ColStats<?> colStats = (ColStats<?>) o;
return colId == colStats.colId
&& Objects.equals(distinctCount, colStats.distinctCount)
- && Objects.equals(serializedMin, colStats.serializedMin)
&& Objects.equals(min, colStats.min)
- && Objects.equals(serializedMax, colStats.serializedMax)
&& Objects.equals(max, colStats.max)
&& Objects.equals(nullCount, colStats.nullCount)
&& Objects.equals(avgLen, colStats.avgLen)
@@ -200,16 +212,7 @@ public class ColStats<T> {
@Override
public int hashCode() {
- return Objects.hash(
- colId,
- distinctCount,
- serializedMin,
- min,
- serializedMax,
- max,
- nullCount,
- avgLen,
- maxLen);
+ return Objects.hash(colId, distinctCount, min, max, nullCount, avgLen, maxLen);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/Stats.java b/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
similarity index 93%
rename from paimon-core/src/main/java/org/apache/paimon/stats/Stats.java
rename to paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
index 046b66a51..60a121cd9 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/Stats.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/Statistics.java
@@ -18,6 +18,7 @@
package org.apache.paimon.stats;
+import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.schema.TableSchema;
@@ -46,7 +47,8 @@ import java.util.OptionalLong;
* <li>colStats: column stats map
* </ul>
*/
-public class Stats {
+@Experimental
+public class Statistics {
// ID of the snapshot this statistics collected from
private static final String FIELD_SNAPSHOT_ID = "snapshotId";
@@ -74,7 +76,7 @@ public class Stats {
private final Map<String, ColStats<?>> colStats;
@JsonCreator
- public Stats(
+ public Statistics(
@JsonProperty(FIELD_SNAPSHOT_ID) long snapshotId,
@JsonProperty(FIELD_SCHEMA_ID) long schemaId,
@JsonProperty(FIELD_MERGED_RECORD_COUNT) @Nullable Long mergedRecordCount,
@@ -87,7 +89,8 @@ public class Stats {
this.colStats = colStats;
}
- public Stats(long snapshotId, long schemaId, Long mergedRecordCount, Long mergedRecordSize) {
+ public Statistics(
+ long snapshotId, long schemaId, Long mergedRecordCount, Long mergedRecordSize) {
this(snapshotId, schemaId, mergedRecordCount, mergedRecordSize, Collections.emptyMap());
}
@@ -161,14 +164,14 @@ public class Stats {
return JsonSerdeUtil.toJson(this);
}
- public static Stats fromJson(String json) {
- return JsonSerdeUtil.fromJson(json, Stats.class);
+ public static Statistics fromJson(String json) {
+ return JsonSerdeUtil.fromJson(json, Statistics.class);
}
- public static Stats fromPath(FileIO fileIO, Path path) {
+ public static Statistics fromPath(FileIO fileIO, Path path) {
try {
String json = fileIO.readFileUtf8(path);
- return Stats.fromJson(json);
+ return Statistics.fromJson(json);
} catch (IOException e) {
throw new RuntimeException("Fails to read snapshot from path " + path, e);
}
@@ -182,7 +185,7 @@ public class Stats {
if (o == null || getClass() != o.getClass()) {
return false;
}
- Stats stats = (Stats) o;
+ Statistics stats = (Statistics) o;
return snapshotId == stats.snapshotId
&& schemaId == stats.schemaId
&& Objects.equals(mergedRecordCount, stats.mergedRecordCount)
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/StatsFile.java b/paimon-core/src/main/java/org/apache/paimon/stats/StatsFile.java
index 252f785bb..95f099f5f 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/StatsFile.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/StatsFile.java
@@ -41,8 +41,8 @@ public class StatsFile {
*
* @return stats
*/
- public Stats read(String fileName) {
- return Stats.fromPath(fileIO, pathFactory.toPath(fileName));
+ public Statistics read(String fileName) {
+ return Statistics.fromPath(fileIO, pathFactory.toPath(fileName));
}
/**
@@ -50,7 +50,7 @@ public class StatsFile {
*
* @return the written file name
*/
- public String write(Stats stats) {
+ public String write(Statistics stats) {
Path path = pathFactory.newPath();
try {
diff --git a/paimon-core/src/main/java/org/apache/paimon/stats/StatsFileHandler.java b/paimon-core/src/main/java/org/apache/paimon/stats/StatsFileHandler.java
index 73ed6c5d4..b240c8e14 100644
--- a/paimon-core/src/main/java/org/apache/paimon/stats/StatsFileHandler.java
+++ b/paimon-core/src/main/java/org/apache/paimon/stats/StatsFileHandler.java
@@ -43,7 +43,7 @@ public class StatsFileHandler {
*
* @return the written file name
*/
- public String writeStats(Stats stats) {
+ public String writeStats(Statistics stats) {
stats.serializeFieldsToString(schemaManager.schema(stats.schemaId()));
return statsFile.write(stats);
}
@@ -53,7 +53,7 @@ public class StatsFileHandler {
*
* @return stats
*/
- public Optional<Stats> readStats() {
+ public Optional<Statistics> readStats() {
Long latestSnapshotId = snapshotManager.latestSnapshotId();
if (latestSnapshotId == null) {
throw new IllegalStateException("Unable to obtain the latest schema");
@@ -66,15 +66,15 @@ public class StatsFileHandler {
*
* @return stats
*/
- public Optional<Stats> readStats(long snapshotId) {
+ public Optional<Statistics> readStats(long snapshotId) {
return readStats(snapshotManager.snapshot(snapshotId));
}
- public Optional<Stats> readStats(Snapshot snapshot) {
+ public Optional<Statistics> readStats(Snapshot snapshot) {
if (snapshot.statistics() == null) {
return Optional.empty();
} else {
- Stats stats = statsFile.read(snapshot.statistics());
+ Statistics stats = statsFile.read(snapshot.statistics());
stats.deserializeFieldsFromString(schemaManager.schema(stats.schemaId()));
return Optional.of(stats);
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
index 5929bc1ab..041470055 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/AbstractFileStoreTable.java
@@ -34,6 +34,7 @@ import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaValidation;
import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.CallbackUtils;
import org.apache.paimon.table.sink.CommitCallback;
import org.apache.paimon.table.sink.DynamicBucketRowKeyExtractor;
@@ -97,6 +98,16 @@ public abstract class AbstractFileStoreTable implements FileStoreTable {
this.catalogEnvironment = catalogEnvironment;
}
+ @Override
+ public Optional<Statistics> statistics() {
+ // todo: support time travel
+ Snapshot latestSnapshot = snapshotManager().latestSnapshot();
+ if (latestSnapshot != null) {
+ return store().newStatsFileHandler().readStats(latestSnapshot);
+ }
+ return Optional.empty();
+ }
+
@Override
public BucketMode bucketMode() {
return store().bucketMode();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
index 66b9142d2..f0d52b641 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/ReadonlyTable.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.InnerTableCommit;
import org.apache.paimon.table.sink.InnerTableWrite;
@@ -47,6 +48,11 @@ public interface ReadonlyTable extends InnerTable {
return Optional.empty();
}
+ @Override
+ default Optional<Statistics> statistics() {
+ return Optional.empty();
+ }
+
@Override
default BatchWriteBuilder newBatchWriteBuilder() {
throw new UnsupportedOperationException(
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/Table.java b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
index b9c28cd69..282a8b09d 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/Table.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/Table.java
@@ -20,6 +20,7 @@ package org.apache.paimon.table;
import org.apache.paimon.annotation.Experimental;
import org.apache.paimon.annotation.Public;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.paimon.table.sink.StreamWriteBuilder;
import org.apache.paimon.table.source.ReadBuilder;
@@ -58,6 +59,10 @@ public interface Table extends Serializable {
/** Optional comment of this table. */
Optional<String> comment();
+ /** Optional statistics of this table. */
+ @Experimental
+ Optional<Statistics> statistics();
+
// ================= Table Operations ====================
/** Copy this table with adding dynamic options. */
diff --git a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 58d961b49..40526b156 100644
--- a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -36,7 +36,7 @@ import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
import org.apache.paimon.stats.ColStats;
-import org.apache.paimon.stats.Stats;
+import org.apache.paimon.stats.Statistics;
import org.apache.paimon.stats.StatsFileHandler;
import org.apache.paimon.testutils.assertj.AssertionUtils;
import org.apache.paimon.types.DataField;
@@ -790,16 +790,16 @@ public class FileStoreCommitTest {
// Analyze and check
HashMap<String, ColStats<?>> fakeColStatsMap = new HashMap<>();
- fakeColStatsMap.put("orderId", new ColStats<>(3, 10L, 1L, 10L, 0L, 8L, 8L));
- Stats fakeStats =
- new Stats(
+ fakeColStatsMap.put("orderId", ColStats.newColStats(3, 10L, 1L, 10L, 0L, 8L, 8L));
+ Statistics fakeStats =
+ new Statistics(
latestSnapshot.id(),
latestSnapshot.schemaId(),
10L,
1000L,
fakeColStatsMap);
fileStoreCommit.commitStatistics(fakeStats, Long.MAX_VALUE);
- Optional<Stats> readStats = statsFileHandler.readStats();
+ Optional<Statistics> readStats = statsFileHandler.readStats();
assertThat(readStats).isPresent();
assertThat(readStats.get()).isEqualTo(fakeStats);
@@ -821,9 +821,9 @@ public class FileStoreCommitTest {
// Then we need to analyze again
latestSnapshot = store.snapshotManager().latestSnapshot();
fakeColStatsMap = new HashMap<>();
- fakeColStatsMap.put("orderId", new ColStats<>(3, 30L, 1L, 30L, 0L, 8L, 8L));
+ fakeColStatsMap.put("orderId", ColStats.newColStats(3, 30L, 1L, 30L, 0L, 8L, 8L));
fakeStats =
- new Stats(
+ new Statistics(
latestSnapshot.id(),
latestSnapshot.schemaId(),
30L,
@@ -836,7 +836,7 @@ public class FileStoreCommitTest {
// Analyze without col stats and check
latestSnapshot = store.snapshotManager().latestSnapshot();
- fakeStats = new Stats(latestSnapshot.id(), latestSnapshot.schemaId(), 30L, 3000L);
+ fakeStats = new Statistics(latestSnapshot.id(), latestSnapshot.schemaId(), 30L, 3000L);
fileStoreCommit.commitStatistics(fakeStats, Long.MAX_VALUE);
readStats = statsFileHandler.readStats();
assertThat(readStats).isPresent();
diff --git a/paimon-core/src/test/java/org/apache/paimon/stats/StatsFileTest.java b/paimon-core/src/test/java/org/apache/paimon/stats/StatsFileTest.java
index 53ea463a1..047aaddbb 100644
--- a/paimon-core/src/test/java/org/apache/paimon/stats/StatsFileTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/stats/StatsFileTest.java
@@ -55,7 +55,7 @@ public class StatsFileTest {
StatsFile file = new StatsFile(LocalFileIO.create(), pathFactory);
HashMap<String, ColStats<?>> colStatsMap = new HashMap<>();
colStatsMap.put("orderId", new ColStats<>(0, 10L, "111", "222", 0L, 8L, 8L));
- Stats stats = new Stats(1L, 0L, 10L, 1000L, colStatsMap);
+ Statistics stats = new Statistics(1L, 0L, 10L, 1000L, colStatsMap);
String fileName = file.write(stats);
assertThat(file.exists(fileName)).isTrue();
diff --git a/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
new file mode 100644
index 000000000..c6f788eab
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.3/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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.paimon.spark.sql
+
+class AnalyzeTableTest extends AnalyzeTableTestBase {}
diff --git a/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
new file mode 100644
index 000000000..c6f788eab
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.4/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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.paimon.spark.sql
+
+class AnalyzeTableTest extends AnalyzeTableTestBase {}
diff --git a/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
new file mode 100644
index 000000000..c6f788eab
--- /dev/null
+++ b/paimon-spark/paimon-spark-3.5/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTest.scala
@@ -0,0 +1,20 @@
+/*
+ * 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.paimon.spark.sql
+
+class AnalyzeTableTest extends AnalyzeTableTestBase {}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
index 472e80694..bec9d26ce 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/catalyst/analysis/PaimonAnalysis.scala
@@ -18,15 +18,15 @@
package org.apache.paimon.spark.catalyst.analysis
import org.apache.paimon.spark.SparkTable
-import org.apache.paimon.spark.catalyst.plans.logical.{PaimonTableValuedFunctions, PaimonTableValueFunction}
-import org.apache.paimon.spark.commands.{PaimonDynamicPartitionOverwriteCommand, PaimonTruncateTableCommand}
+import org.apache.paimon.spark.catalyst.analysis.PaimonRelation.isPaimonTable
+import org.apache.paimon.spark.commands.{PaimonAnalyzeTableColumnCommand, PaimonDynamicPartitionOverwriteCommand, PaimonTruncateTableCommand}
import org.apache.paimon.table.FileStoreTable
-import PaimonRelation.isPaimonTable
import org.apache.spark.sql.SparkSession
-import org.apache.spark.sql.catalyst.analysis.ResolvedPartitionSpec
-import org.apache.spark.sql.catalyst.plans.logical.{LogicalPlan, MergeIntoTable, OverwritePartitionsDynamic, TruncatePartition, TruncateTable}
+import org.apache.spark.sql.catalyst.analysis.ResolvedTable
+import org.apache.spark.sql.catalyst.plans.logical._
import org.apache.spark.sql.catalyst.rules.Rule
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
class PaimonAnalysis(session: SparkSession) extends Rule[LogicalPlan] {
@@ -49,6 +49,29 @@ case class PaimonPostHocResolutionRules(session: SparkSession) extends Rule[Logi
case t @ TruncateTable(PaimonRelation(table)) if t.resolved =>
PaimonTruncateTableCommand(table, Map.empty)
+ case a @ AnalyzeTable(
+ ResolvedTable(catalog, identifier, table: SparkTable, _),
+ partitionSpec,
+ noScan) if a.resolved =>
+ if (partitionSpec.nonEmpty) {
+ throw new UnsupportedOperationException("Analyze table partition is not supported")
+ } else if (noScan) {
+ throw new IllegalArgumentException("NOSCAN is ineffective with paimon")
+ } else {
+ PaimonAnalyzeTableColumnCommand(
+ catalog,
+ identifier,
+ table,
+ Option.apply(Seq()),
+ allColumns = false)
+ }
+
+ case a @ AnalyzeColumn(
+ ResolvedTable(catalog, identifier, table: SparkTable, _),
+ columnNames,
+ allColumns) if a.resolved =>
+ PaimonAnalyzeTableColumnCommand(catalog, identifier, table, columnNames, allColumns)
+
case _ => plan
}
}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
new file mode 100644
index 000000000..e9355a5cb
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonAnalyzeTableColumnCommand.scala
@@ -0,0 +1,165 @@
+/*
+ * 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.paimon.spark.commands
+
+import org.apache.paimon.schema.TableSchema
+import org.apache.paimon.spark.SparkTable
+import org.apache.paimon.spark.leafnode.PaimonLeafRunnableCommand
+import org.apache.paimon.stats.{ColStats, Statistics}
+import org.apache.paimon.table.FileStoreTable
+import org.apache.paimon.table.sink.BatchWriteBuilder
+
+import org.apache.parquet.Preconditions
+import org.apache.spark.sql.{Row, SparkSession, Utils}
+import org.apache.spark.sql.catalyst.expressions.Attribute
+import org.apache.spark.sql.catalyst.plans.logical.ColumnStat
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog}
+import org.apache.spark.sql.execution.datasources.v2.DataSourceV2Relation
+import org.apache.spark.sql.types.{DataType, Decimal, DecimalType, TimestampType}
+
+import java.util
+import java.util.UUID
+
+import scala.collection.JavaConverters._
+
+/** Command for analyze table and column. */
+case class PaimonAnalyzeTableColumnCommand(
+ catalog: TableCatalog,
+ identifier: Identifier,
+ v2Table: SparkTable,
+ columnNames: Option[Seq[String]],
+ allColumns: Boolean)
+ extends PaimonLeafRunnableCommand
+ with WithFileStoreTable {
+
+ override def table: FileStoreTable = v2Table.getTable.asInstanceOf[FileStoreTable]
+
+ override def run(sparkSession: SparkSession): Seq[Row] = {
+ val relation = DataSourceV2Relation.create(v2Table, Some(catalog), Some(identifier))
+ val currentSnapshot = table.snapshotManager().latestSnapshot()
+
+ // compute stats
+ val attributes = getColumnsToAnalyze(relation, columnNames, allColumns)
+ val totalSize = Utils.calculateTotalSize(
+ sparkSession.sessionState,
+ table.name(),
+ Some(table.location().toUri))
+ val (mergedRecordCount, colStats) = Utils.computeColumnStats(sparkSession, relation, attributes)
+
+ val totalRecordCount = currentSnapshot.totalRecordCount()
+ Preconditions.checkState(
+ totalRecordCount >= mergedRecordCount,
+ s"totalRecordCount: $totalRecordCount should be greater or equal than mergedRecordCount: $mergedRecordCount.")
+ val mergedRecordSize = totalSize * (mergedRecordCount.toDouble / totalRecordCount).toLong
+
+ // convert to paimon stats
+ val tableSchema = table.schema()
+
+ val colStatsMap: util.Map[String, ColStats[_]] = new util.HashMap
+ for ((attr, stats) <- colStats) {
+ colStatsMap.put(attr.name, toPaimonColStats(attr, stats, tableSchema))
+ }
+
+ val stats = new Statistics(
+ currentSnapshot.id(),
+ currentSnapshot.schemaId(),
+ mergedRecordCount,
+ mergedRecordSize,
+ colStatsMap)
+
+ // commit stats
+ val commit = table.store.newCommit(UUID.randomUUID.toString)
+ commit.commitStatistics(stats, BatchWriteBuilder.COMMIT_IDENTIFIER)
+
+ Seq.empty[Row]
+ }
+
+ private def getColumnsToAnalyze(
+ relation: DataSourceV2Relation,
+ columnNames: Option[Seq[String]],
+ allColumns: Boolean): Seq[Attribute] = {
+ if (columnNames.isDefined && allColumns) {
+ throw new UnsupportedOperationException(
+ "Parameter `columnNames` and `allColumns` are " +
+ "mutually exclusive. Only one of them should be specified.")
+ }
+ val columnsToAnalyze = if (allColumns) {
+ relation.output
+ } else {
+ columnNames.get.map {
+ col =>
+ val exprOption = relation.output.find(attr => conf.resolver(attr.name, col))
+ exprOption.getOrElse(
+ throw new RuntimeException(s"Column $col not found in ${relation.table.name()}."))
+ }
+ }
+ columnsToAnalyze.foreach {
+ attr =>
+ if (!Utils.analyzeSupportsType(attr.dataType)) {
+ throw new UnsupportedOperationException(
+ s"Analyzing on col: ${attr.name}, data type: ${attr.dataType} is not supported.")
+ }
+ }
+ columnsToAnalyze
+ }
+
+ private def toPaimonColStats(
+ attribute: Attribute,
+ columnStat: ColumnStat,
+ paimonSchema: TableSchema): ColStats[_] = {
+ paimonSchema.fields.asScala
+ .find(_.name() == attribute.name)
+ .map {
+ field =>
+ ColStats.newColStats(
+ field.id(),
+ if (columnStat.distinctCount.isDefined) columnStat.distinctCount.get.longValue
+ else null,
+ if (columnStat.min.isDefined)
+ toPaimonData(columnStat.min.get, attribute.dataType).asInstanceOf[Comparable[Any]]
+ else null,
+ if (columnStat.max.isDefined)
+ toPaimonData(columnStat.max.get, attribute.dataType).asInstanceOf[Comparable[Any]]
+ else null,
+ if (columnStat.nullCount.isDefined) columnStat.nullCount.get.longValue else null,
+ if (columnStat.avgLen.isDefined) columnStat.avgLen.get else null,
+ if (columnStat.maxLen.isDefined) columnStat.maxLen.get else null
+ )
+ }
+ .getOrElse(throw new RuntimeException(s"Column ${attribute.name} is not found in schema."))
+ }
+
+ /**
+ * Convert data from spark type to paimon, only cover datatype meet [[Utils.hasMinMax]] currently.
+ */
+ private def toPaimonData(o: Any, dataType: DataType): Any = {
+ dataType match {
+ case d if !Utils.hasMinMax(d) =>
+ // should not reach here
+ throw new UnsupportedOperationException(s"Unsupported data type $d, value is $o.")
+ case _: DecimalType =>
+ val d = o.asInstanceOf[Decimal]
+ org.apache.paimon.data.Decimal.fromBigDecimal(d.toJavaBigDecimal, d.precision, d.scale)
+ case _: TimestampType =>
+ val l = o.asInstanceOf[Long]
+ org.apache.paimon.data.Timestamp.fromSQLTimestamp(DateTimeUtils.toJavaTimestamp(l))
+ case _ => o
+ }
+ }
+}
diff --git a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
index e8f1b418a..5ff1de73c 100644
--- a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
+++ b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/Utils.scala
@@ -17,13 +17,20 @@
*/
package org.apache.spark.sql
+import org.apache.spark.sql.catalyst.TableIdentifier
+import org.apache.spark.sql.catalyst.catalog.{CatalogTable, CatalogUtils}
import org.apache.spark.sql.catalyst.expressions.{Attribute, Expression}
-import org.apache.spark.sql.catalyst.plans.logical.LogicalPlan
+import org.apache.spark.sql.catalyst.plans.logical.{ColumnStat, LogicalPlan}
import org.apache.spark.sql.connector.expressions.{FieldReference, NamedReference}
+import org.apache.spark.sql.execution.command.CommandUtils
import org.apache.spark.sql.execution.datasources.DataSourceStrategy
+import org.apache.spark.sql.internal.SessionState
import org.apache.spark.sql.sources.Filter
+import org.apache.spark.sql.types.{BinaryType, BooleanType, DataType, DatetimeType, DecimalType, DoubleType, FloatType, IntegralType, StringType}
import org.apache.spark.util.{Utils => SparkUtils}
+import java.net.URI
+
/**
* Some classes or methods defined in the spark project are marked as private under
* [[org.apache.spark.sql]] package, Hence, use this class to adapt then so that we can use them
@@ -68,4 +75,42 @@ object Utils {
def bytesToString(size: Long): String = {
SparkUtils.bytesToString(size)
}
+
+ // for analyze
+ def calculateTotalSize(
+ sessionState: SessionState,
+ tableName: String,
+ locationUri: Option[URI]): Long = {
+ CommandUtils.calculateSingleLocationSize(
+ sessionState,
+ new TableIdentifier(tableName),
+ locationUri)
+ }
+
+ def computeColumnStats(
+ sparkSession: SparkSession,
+ relation: LogicalPlan,
+ columns: Seq[Attribute]): (Long, Map[Attribute, ColumnStat]) = {
+ CommandUtils.computeColumnStats(sparkSession, relation, columns)
+ }
+
+ /** [[IntegralType]] is private in spark, therefore we need add it here. */
+ def analyzeSupportsType(dataType: DataType): Boolean = dataType match {
+ case _: IntegralType => true
+ case _: DecimalType => true
+ case DoubleType | FloatType => true
+ case BooleanType => true
+ case _: DatetimeType => true
+ case BinaryType | StringType => true
+ case _ => false
+ }
+
+ def hasMinMax(dataType: DataType): Boolean = dataType match {
+ case _: IntegralType => true
+ case _: DecimalType => true
+ case DoubleType | FloatType => true
+ case BooleanType => true
+ case _: DatetimeType => true
+ case _ => false
+ }
}
diff --git a/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
new file mode 100644
index 000000000..d0f4093e4
--- /dev/null
+++ b/paimon-spark/paimon-spark-common/src/test/scala/org/apache/paimon/spark/sql/AnalyzeTableTestBase.scala
@@ -0,0 +1,247 @@
+/*
+ * 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.paimon.spark.sql
+
+import org.apache.paimon.data.Decimal
+import org.apache.paimon.spark.PaimonSparkTestBase
+import org.apache.paimon.stats.ColStats
+import org.apache.paimon.utils.DateTimeUtils
+
+import org.apache.spark.sql.Row
+import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.jupiter.api.Assertions
+
+abstract class AnalyzeTableTestBase extends PaimonSparkTestBase {
+
+ test("Paimon analyze: analyze table only") {
+ spark.sql(s"""
+ |CREATE TABLE T (id STRING, name STRING, i INT, l LONG)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO T VALUES ('1', 'a', 1, 1)")
+ spark.sql(s"INSERT INTO T VALUES ('2', 'aaa', 1, 2)")
+
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS")
+
+ val stats = loadTable("T").statistics().get()
+ Assertions.assertEquals(2L, stats.mergedRecordCount().getAsLong)
+ Assertions.assertTrue(stats.mergedRecordSize().isPresent)
+ Assertions.assertTrue(stats.colStats().isEmpty)
+ }
+
+ test("Paimon analyze: analyze no scan") {
+ spark.sql(s"CREATE TABLE T (id STRING, name STRING)")
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS NOSCAN"))
+ .hasMessageContaining("NOSCAN is ineffective with paimon")
+ }
+
+ test("Paimon analyze: analyze table partition") {
+ spark.sql(s"CREATE TABLE T (id STRING, name STRING, pt STRING) PARTITIONED BY (id, pt) ")
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T PARTITION (pt = '1') COMPUTE STATISTICS"))
+ .hasMessageContaining("Analyze table partition is not supported")
+ }
+
+ test("Paimon analyze: analyze all supported cols") {
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id STRING, name STRING, byte_col BYTE, short_col SHORT, int_col INT, long_col LONG,
+ |float_col FLOAT, double_col DOUBLE, decimal_col DECIMAL(10, 5), boolean_col BOOLEAN, date_col DATE,
+ |timestamp_col TIMESTAMP, binary BINARY)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ spark.sql(
+ s"INSERT INTO T VALUES ('1', 'a', 1, 1, 1, 1, 1.0, 1.0, 12.12345, true, cast('2020-01-01' as date), cast('2020-01-01 00:00:00' as timestamp), binary('example binary1'))")
+ spark.sql(
+ s"INSERT INTO T VALUES ('2', 'aaa', 1, null, 1, 1, 1.0, 1.0, 12.12345, true, cast('2020-01-02' as date), cast('2020-01-02 00:00:00' as timestamp), binary('example binary1'))")
+ spark.sql(
+ s"INSERT INTO T VALUES ('3', 'bbbb', 2, 1, 1, 1, 1.0, 1.0, 22.12345, true, cast('2020-01-02' as date), cast('2020-01-02 00:00:00' as timestamp), null)")
+ spark.sql(
+ s"INSERT INTO T VALUES ('4', 'bbbbbbbb', 2, 2, 2, 2, 2.0, 2.0, 22.12345, false, cast('2020-01-01' as date), cast('2020-01-01 00:00:00' as timestamp), binary('example binary2'))")
+
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR ALL COLUMNS")
+
+ var stats = loadTable("T").statistics().get()
+ Assertions.assertEquals(4L, stats.mergedRecordCount().getAsLong)
+ Assertions.assertTrue(stats.mergedRecordSize().isPresent)
+
+ var colStats = stats.colStats()
+ Assertions.assertEquals(ColStats.newColStats(0, 4, null, null, 0, 1, 1), colStats.get("id"))
+ Assertions.assertEquals(ColStats.newColStats(1, 4, null, null, 0, 4, 8), colStats.get("name"))
+ Assertions.assertEquals(
+ ColStats.newColStats(2, 2, 1.toByte, 2.toByte, 0, 1, 1),
+ colStats.get("byte_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(3, 2, 1.toShort, 2.toShort, 1, 2, 2),
+ colStats.get("short_col"))
+ Assertions.assertEquals(ColStats.newColStats(4, 2, 1, 2, 0, 4, 4), colStats.get("int_col"))
+ Assertions.assertEquals(ColStats.newColStats(5, 2, 1L, 2L, 0, 8, 8), colStats.get("long_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(6, 2, 1.0f, 2.0f, 0, 4, 4),
+ colStats.get("float_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(7, 2, 1.0d, 2.0d, 0, 8, 8),
+ colStats.get("double_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(
+ 8,
+ 2,
+ Decimal.fromBigDecimal(new java.math.BigDecimal("12.12345"), 10, 5),
+ Decimal.fromBigDecimal(new java.math.BigDecimal("22.12345"), 10, 5),
+ 0,
+ 8,
+ 8),
+ colStats.get("decimal_col")
+ )
+ Assertions.assertEquals(
+ ColStats.newColStats(9, 2, false, true, 0, 1, 1),
+ colStats.get("boolean_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(10, 2, 18262, 18263, 0, 4, 4),
+ colStats.get("date_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(
+ 11,
+ 2,
+ DateTimeUtils.parseTimestampData("2020-01-01 00:00:00", 0),
+ DateTimeUtils.parseTimestampData("2020-01-02 00:00:00", 0),
+ 0,
+ 8,
+ 8),
+ colStats.get("timestamp_col")
+ )
+ Assertions.assertEquals(
+ ColStats.newColStats(12, 2, null, null, 1, 15, 15),
+ colStats.get("binary"))
+
+ spark.sql(
+ s"INSERT INTO T VALUES ('5', 'bbbbbbbbbbbbbbbb', 3, 3, 3, 3, 3.0, 3.0, 32.12345, false, cast('2020-01-03' as date), cast('2020-01-03 00:00:00' as timestamp), binary('binary3'))")
+
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR ALL COLUMNS")
+
+ stats = loadTable("T").statistics().get()
+ Assertions.assertEquals(5L, stats.mergedRecordCount().getAsLong)
+ Assertions.assertTrue(stats.mergedRecordSize().isPresent)
+
+ colStats = stats.colStats()
+ Assertions.assertEquals(ColStats.newColStats(0, 5, null, null, 0, 1, 1), colStats.get("id"))
+ Assertions.assertEquals(ColStats.newColStats(1, 5, null, null, 0, 7, 16), colStats.get("name"))
+ Assertions.assertEquals(
+ ColStats.newColStats(2, 3, 1.toByte, 3.toByte, 0, 1, 1),
+ colStats.get("byte_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(3, 3, 1.toShort, 3.toShort, 1, 2, 2),
+ colStats.get("short_col"))
+ Assertions.assertEquals(ColStats.newColStats(4, 3, 1, 3, 0, 4, 4), colStats.get("int_col"))
+ Assertions.assertEquals(ColStats.newColStats(5, 3, 1L, 3L, 0, 8, 8), colStats.get("long_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(6, 3, 1.0f, 3.0f, 0, 4, 4),
+ colStats.get("float_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(7, 3, 1.0d, 3.0d, 0, 8, 8),
+ colStats.get("double_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(
+ 8,
+ 3,
+ Decimal.fromBigDecimal(new java.math.BigDecimal("12.12345"), 10, 5),
+ Decimal.fromBigDecimal(new java.math.BigDecimal("32.12345"), 10, 5),
+ 0,
+ 8,
+ 8),
+ colStats.get("decimal_col")
+ )
+ Assertions.assertEquals(
+ ColStats.newColStats(9, 2, false, true, 0, 1, 1),
+ colStats.get("boolean_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(10, 3, 18262, 18264, 0, 4, 4),
+ colStats.get("date_col"))
+ Assertions.assertEquals(
+ ColStats.newColStats(
+ 11,
+ 3,
+ DateTimeUtils.parseTimestampData("2020-01-01 00:00:00", 0),
+ DateTimeUtils.parseTimestampData("2020-01-03 00:00:00", 0),
+ 0,
+ 8,
+ 8),
+ colStats.get("timestamp_col")
+ )
+ Assertions.assertEquals(
+ ColStats.newColStats(12, 3, null, null, 1, 13, 15),
+ colStats.get("binary"))
+ }
+
+ test("Paimon analyze: analyze unsupported cols") {
+ spark.sql(
+ s"""
+ |CREATE TABLE T (id STRING, m MAP<INT, STRING>, l ARRAY<INT>, s STRUCT<i:INT, s:STRING>)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO T VALUES ('1', map(1, 'a'), array(1), struct(1, 'a'))")
+
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS m"))
+ .hasMessageContaining("not supported")
+
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS l"))
+ .hasMessageContaining("not supported")
+
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS s"))
+ .hasMessageContaining("not supported")
+ }
+
+ test("Paimon analyze: analyze non-exist cols") {
+ spark.sql(s"""
+ |CREATE TABLE T (id STRING, name STRING, i INT, l LONG)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ assertThatThrownBy(() => spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS fake_col"))
+ .hasMessageContaining("not found")
+ }
+
+ test("Paimon analyze: analyze specialized cols") {
+ spark.sql(s"""
+ |CREATE TABLE T (id STRING, name STRING, i INT, l LONG)
+ |USING PAIMON
+ |TBLPROPERTIES ('primary-key'='id')
+ |""".stripMargin)
+
+ spark.sql(s"INSERT INTO T VALUES ('1', 'a', 1, 1)")
+ spark.sql(s"INSERT INTO T VALUES ('2', 'aaa', 1, 2)")
+
+ spark.sql(s"ANALYZE TABLE T COMPUTE STATISTICS FOR COLUMNS name, i")
+
+ val colStats = loadTable("T").statistics().get().colStats()
+ Assertions.assertEquals(null, colStats.get("id"))
+ Assertions.assertEquals(ColStats.newColStats(1, 2, null, null, 0, 2, 3), colStats.get("name"))
+ Assertions.assertEquals(ColStats.newColStats(2, 1, 1, 1, 0, 4, 4), colStats.get("i"))
+ Assertions.assertEquals(null, colStats.get("l"))
+
+ checkAnswer(
+ spark.sql(s"SELECT * from T ORDER BY id"),
+ Row("1", "a", 1, 1) :: Row("2", "aaa", 1, 2) :: Nil)
+ }
+}