You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "JingsongLi (via GitHub)" <gi...@apache.org> on 2026/05/20 09:57:15 UTC
[PR] [WIP][format] Add paimon-mosaic-format module with reader and writer [paimon]
JingsongLi opened a new pull request, #7917:
URL: https://github.com/apache/paimon/pull/7917
### Purpose
See https://paimon.apache.org/docs/mosaic/
Introduces the Mosaic file format integration for Paimon with:
- MosaicRecordsReader: row-group level predicate filtering using statistics, column projection, and correct returnedPosition tracking
- MosaicRecordsWriter: BundleFormatWriter with writerMetadata() support for in-memory stats capture (avoids re-reading files on object stores)
- MosaicSimpleStatsExtractor: stats extraction from file or writerMetadata, with SimpleColStatsCollector integration
- MosaicObjects: byte[] to Paimon object conversion for all supported types
- Comprehensive test suite (6 test classes covering unit and integration tests)
### Tests
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #7917:
URL: https://github.com/apache/paimon/pull/7917
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3278436360
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicWriterFactory.java:
##########
@@ -0,0 +1,63 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.types.RowType;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** A factory to create Mosaic {@link FormatWriter}. */
+public class MosaicWriterFactory implements FormatWriterFactory {
+
+ private final RowType rowType;
+ private final FileFormatFactory.FormatContext formatContext;
+ private final List<String> statsColumnNames;
+ private final @Nullable Integer numBuckets;
+
+ public MosaicWriterFactory(RowType rowType, FileFormatFactory.FormatContext formatContext) {
+ this.rowType = rowType;
+ this.formatContext = formatContext;
+ String statsColumnsValue = formatContext.options().get(MosaicFileFormat.STATS_COLUMNS);
+ if (statsColumnsValue == null || statsColumnsValue.trim().isEmpty()) {
+ this.statsColumnNames = new ArrayList<>();
+ } else {
+ this.statsColumnNames =
+ Arrays.stream(statsColumnsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ }
+ this.numBuckets = formatContext.options().get(MosaicFileFormat.NUM_BUCKETS);
+ }
+
+ @Override
+ public FormatWriter create(PositionOutputStream out, String compression) {
+ // only support zstd, ignore compression
Review Comment:
If Mosaic only supports ZSTD for now, silently ignoring this argument is still not safe. Users can configure `file.compression = none/lz4/snappy/...`, and Paimon may also put the configured compression into file names when `file.suffix.include.compression` is enabled, while the actual Mosaic footer remains ZSTD. Please fail fast unless the value is null or `zstd` (case-insensitive), or wait for the Mosaic Java API to expose and wire other supported compression constants explicitly.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3274635895
##########
paimon-mosaic/pom.xml:
##########
@@ -0,0 +1,86 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <parent>
+ <artifactId>paimon-parent</artifactId>
+ <groupId>org.apache.paimon</groupId>
+ <version>1.5-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>paimon-mosaic</artifactId>
+ <name>Paimon : Mosaic Format</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.paimon</groupId>
+ <artifactId>mosaic</artifactId>
+ <version>0.1.0-SNAPSHOT</version>
Review Comment:
This still makes the default Paimon reactor depend on an external SNAPSHOT artifact. Since `paimon-mosaic` is listed in the root `<modules>`, a clean `mvn -pl paimon-mosaic -am ... compile` fails with `Could not find artifact org.apache.paimon:mosaic:jar:0.1.0-SNAPSHOT`. Please use a released artifact, build/include this dependency in the same releaseable reactor, or keep this module out of the default root reactor/profile-gated until the Mosaic Java artifact is released.
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.mosaic.ColumnStatistics;
+import org.apache.paimon.mosaic.MosaicReader;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+
+import javax.annotation.Nullable;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import static org.apache.paimon.format.mosaic.MosaicObjects.convertStatsValue;
+
+/** Extracts statistics from Mosaic file metadata. */
+public class MosaicSimpleStatsExtractor implements SimpleStatsExtractor {
+
+ private final RowType rowType;
+ private final SimpleColStatsCollector.Factory[] statsCollectors;
+
+ public MosaicSimpleStatsExtractor(
+ RowType rowType, SimpleColStatsCollector.Factory[] statsCollectors) {
+ this.rowType = rowType;
+ this.statsCollectors = statsCollectors;
+ }
+
+ @Override
+ public SimpleColStats[] extract(FileIO fileIO, Path path, long length) {
+ MosaicInputFileAdapter inputFile = new MosaicInputFileAdapter(fileIO, path);
+ try (BufferAllocator allocator = new RootAllocator();
+ MosaicReader reader = MosaicReader.open(inputFile, length, allocator)) {
+ return extractFromStats(reader.numRowGroups(), reader::getRowGroupStatistics, null);
Review Comment:
The file-only extraction path loses the configured stats-column set. For a file written with only a subset of `mosaic.stats-columns` (or the default empty set), columns missing from Mosaic's `ColumnStatistics` keep `nullCounts[colIdx] == 0`, so untracked columns are returned as `SimpleColStats(null, null, 0)` instead of `SimpleColStats.NONE`. That can persist an exact zero-null count on fallback/migration paths. Please track which columns actually appear in the Mosaic stats and return `NONE` for unseen columns, or persist/recover the collected stats-column set from the file metadata. The same issue applies to `extractWithFileInfo`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3278243898
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.mosaic.ColumnStatistics;
+import org.apache.paimon.mosaic.MosaicReader;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.format.mosaic.MosaicObjects.convertStatsValue;
+
+/** Extracts statistics from Mosaic file metadata. */
+public class MosaicSimpleStatsExtractor implements SimpleStatsExtractor {
+
+ private final RowType rowType;
+ private final SimpleColStatsCollector.Factory[] statsCollectors;
+
+ public MosaicSimpleStatsExtractor(
+ RowType rowType, SimpleColStatsCollector.Factory[] statsCollectors) {
+ this.rowType = rowType;
+ this.statsCollectors = statsCollectors;
+ }
+
+ @Override
+ public SimpleColStats[] extract(FileIO fileIO, Path path, long length) {
+ try (MosaicInputFileAdapter inputFile = new MosaicInputFileAdapter(fileIO, path);
+ BufferAllocator allocator = new RootAllocator();
+ MosaicReader reader = MosaicReader.open(inputFile, length, allocator)) {
+ return extractFromStats(reader.numRowGroups(), reader::getRowGroupStatistics, null);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to extract stats from " + path, e);
+ }
+ }
+
+ @Override
+ public SimpleColStats[] extract(
+ FileIO fileIO, Path path, long length, @Nullable Object writerMetadata) {
+ if (writerMetadata instanceof MosaicWriterMetadata) {
+ MosaicWriterMetadata meta = (MosaicWriterMetadata) writerMetadata;
+ Set<Integer> statsFieldIndices = resolveStatsFieldIndices(meta.statsColumnNames());
+ return extractFromStats(
+ meta.numRowGroups(), meta::getRowGroupStatistics, statsFieldIndices);
+ }
+ return extract(fileIO, path, length);
+ }
+
+ @Override
+ public Pair<SimpleColStats[], FileInfo> extractWithFileInfo(
+ FileIO fileIO, Path path, long length) {
+ try (MosaicInputFileAdapter inputFile = new MosaicInputFileAdapter(fileIO, path);
+ BufferAllocator allocator = new RootAllocator();
+ MosaicReader reader = MosaicReader.open(inputFile, length, allocator)) {
+ int numRowGroups = reader.numRowGroups();
+ SimpleColStats[] stats =
+ extractFromStats(numRowGroups, reader::getRowGroupStatistics, null);
+ long rowCount = 0;
+ for (int rg = 0; rg < numRowGroups; rg++) {
+ rowCount += reader.rowGroupNumRows(rg);
+ }
+ return Pair.of(stats, new FileInfo(rowCount));
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to extract stats from " + path, e);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private SimpleColStats[] extractFromStats(
+ int numRowGroups,
+ RowGroupStatsProvider statsProvider,
+ @Nullable Set<Integer> statsFieldIndices) {
+ int fieldCount = rowType.getFieldCount();
+ List<String> fieldNames = rowType.getFieldNames();
+ Object[] minValues = new Object[fieldCount];
+ Object[] maxValues = new Object[fieldCount];
+ long[] nullCounts = new long[fieldCount];
+ Set<Integer> seenColumns = new HashSet<>();
+
+ for (int rg = 0; rg < numRowGroups; rg++) {
+ Map<String, ColumnStatistics> statsMap = statsProvider.getRowGroupStatistics(rg);
+ for (Map.Entry<String, ColumnStatistics> entry : statsMap.entrySet()) {
+ int colIdx = fieldNames.indexOf(entry.getKey());
+ if (colIdx < 0) {
+ continue;
+ }
+
+ ColumnStatistics stat = entry.getValue();
+ seenColumns.add(colIdx);
+ nullCounts[colIdx] += stat.getNullCount();
+
+ if (stat.hasMinMax()) {
+ DataType dataType = rowType.getFields().get(colIdx).type();
+ Object min = convertStatsValue(stat.getMin(), dataType);
+ Object max = convertStatsValue(stat.getMax(), dataType);
+ if (min != null) {
+ if (minValues[colIdx] == null) {
+ minValues[colIdx] = min;
+ } else {
+ if (((Comparable<Object>) min).compareTo(minValues[colIdx]) < 0) {
Review Comment:
`MosaicObjects.convertStatsValue` returns `byte[]` for BINARY/VARBINARY stats, but this aggregation path assumes every min/max value is `Comparable`. If a binary stats column spans more than one row group, the second row group will hit `ClassCastException` here. Please either compare through Paimon's byte-array-aware comparator (for example `CompareUtils.compareLiteral(dataType, ...)`) or do not expose min/max for binary columns, and add a multi-row-group binary stats test.
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicWriterFactory.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.paimon.format.mosaic;
+
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.types.RowType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** A factory to create Mosaic {@link FormatWriter}. */
+public class MosaicWriterFactory implements FormatWriterFactory {
+
+ private final RowType rowType;
+ private final FileFormatFactory.FormatContext formatContext;
+ private final List<String> statsColumnNames;
+ private final int numBuckets;
+
+ public MosaicWriterFactory(RowType rowType, FileFormatFactory.FormatContext formatContext) {
+ this.rowType = rowType;
+ this.formatContext = formatContext;
+ String statsColumnsValue = formatContext.options().get(MosaicFileFormat.STATS_COLUMNS);
+ if (statsColumnsValue == null || statsColumnsValue.trim().isEmpty()) {
+ this.statsColumnNames = new ArrayList<>();
+ } else {
+ this.statsColumnNames =
+ Arrays.stream(statsColumnsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ }
+ this.numBuckets = formatContext.options().get(MosaicFileFormat.NUM_BUCKETS);
+ }
+
+ @Override
+ public FormatWriter create(PositionOutputStream out, String compression) throws IOException {
Review Comment:
The `compression` argument from Paimon's writer path is currently ignored. `MosaicRecordsWriter` always uses `WriterOptions`' default compression (zstd) and only forwards the zstd level, so table options like `file.compression = none` or `file.compression.per.level` silently have no effect for Mosaic files. Please map supported Paimon compression values to Mosaic `WriterOptions.compression(...)` (or reject unsupported ones explicitly) and add coverage for at least `zstd` and `none`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3278243898
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicSimpleStatsExtractor.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.format.SimpleStatsExtractor;
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.mosaic.ColumnStatistics;
+import org.apache.paimon.mosaic.MosaicReader;
+import org.apache.paimon.statistics.SimpleColStatsCollector;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Pair;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.apache.paimon.format.mosaic.MosaicObjects.convertStatsValue;
+
+/** Extracts statistics from Mosaic file metadata. */
+public class MosaicSimpleStatsExtractor implements SimpleStatsExtractor {
+
+ private final RowType rowType;
+ private final SimpleColStatsCollector.Factory[] statsCollectors;
+
+ public MosaicSimpleStatsExtractor(
+ RowType rowType, SimpleColStatsCollector.Factory[] statsCollectors) {
+ this.rowType = rowType;
+ this.statsCollectors = statsCollectors;
+ }
+
+ @Override
+ public SimpleColStats[] extract(FileIO fileIO, Path path, long length) {
+ try (MosaicInputFileAdapter inputFile = new MosaicInputFileAdapter(fileIO, path);
+ BufferAllocator allocator = new RootAllocator();
+ MosaicReader reader = MosaicReader.open(inputFile, length, allocator)) {
+ return extractFromStats(reader.numRowGroups(), reader::getRowGroupStatistics, null);
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to extract stats from " + path, e);
+ }
+ }
+
+ @Override
+ public SimpleColStats[] extract(
+ FileIO fileIO, Path path, long length, @Nullable Object writerMetadata) {
+ if (writerMetadata instanceof MosaicWriterMetadata) {
+ MosaicWriterMetadata meta = (MosaicWriterMetadata) writerMetadata;
+ Set<Integer> statsFieldIndices = resolveStatsFieldIndices(meta.statsColumnNames());
+ return extractFromStats(
+ meta.numRowGroups(), meta::getRowGroupStatistics, statsFieldIndices);
+ }
+ return extract(fileIO, path, length);
+ }
+
+ @Override
+ public Pair<SimpleColStats[], FileInfo> extractWithFileInfo(
+ FileIO fileIO, Path path, long length) {
+ try (MosaicInputFileAdapter inputFile = new MosaicInputFileAdapter(fileIO, path);
+ BufferAllocator allocator = new RootAllocator();
+ MosaicReader reader = MosaicReader.open(inputFile, length, allocator)) {
+ int numRowGroups = reader.numRowGroups();
+ SimpleColStats[] stats =
+ extractFromStats(numRowGroups, reader::getRowGroupStatistics, null);
+ long rowCount = 0;
+ for (int rg = 0; rg < numRowGroups; rg++) {
+ rowCount += reader.rowGroupNumRows(rg);
+ }
+ return Pair.of(stats, new FileInfo(rowCount));
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to extract stats from " + path, e);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private SimpleColStats[] extractFromStats(
+ int numRowGroups,
+ RowGroupStatsProvider statsProvider,
+ @Nullable Set<Integer> statsFieldIndices) {
+ int fieldCount = rowType.getFieldCount();
+ List<String> fieldNames = rowType.getFieldNames();
+ Object[] minValues = new Object[fieldCount];
+ Object[] maxValues = new Object[fieldCount];
+ long[] nullCounts = new long[fieldCount];
+ Set<Integer> seenColumns = new HashSet<>();
+
+ for (int rg = 0; rg < numRowGroups; rg++) {
+ Map<String, ColumnStatistics> statsMap = statsProvider.getRowGroupStatistics(rg);
+ for (Map.Entry<String, ColumnStatistics> entry : statsMap.entrySet()) {
+ int colIdx = fieldNames.indexOf(entry.getKey());
+ if (colIdx < 0) {
+ continue;
+ }
+
+ ColumnStatistics stat = entry.getValue();
+ seenColumns.add(colIdx);
+ nullCounts[colIdx] += stat.getNullCount();
+
+ if (stat.hasMinMax()) {
+ DataType dataType = rowType.getFields().get(colIdx).type();
+ Object min = convertStatsValue(stat.getMin(), dataType);
+ Object max = convertStatsValue(stat.getMax(), dataType);
+ if (min != null) {
+ if (minValues[colIdx] == null) {
+ minValues[colIdx] = min;
+ } else {
+ if (((Comparable<Object>) min).compareTo(minValues[colIdx]) < 0) {
Review Comment:
returns for BINARY/VARBINARY stats, but this aggregation path assumes every min/max value is . If a binary stats column spans more than one row group, the second row group will hit here. Please either compare through Paimon's byte-array-aware comparator (for example ) or do not expose min/max for binary columns, and add a multi-row-group binary stats test.
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicWriterFactory.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.paimon.format.mosaic;
+
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.format.FormatWriter;
+import org.apache.paimon.format.FormatWriterFactory;
+import org.apache.paimon.fs.PositionOutputStream;
+import org.apache.paimon.types.RowType;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/** A factory to create Mosaic {@link FormatWriter}. */
+public class MosaicWriterFactory implements FormatWriterFactory {
+
+ private final RowType rowType;
+ private final FileFormatFactory.FormatContext formatContext;
+ private final List<String> statsColumnNames;
+ private final int numBuckets;
+
+ public MosaicWriterFactory(RowType rowType, FileFormatFactory.FormatContext formatContext) {
+ this.rowType = rowType;
+ this.formatContext = formatContext;
+ String statsColumnsValue = formatContext.options().get(MosaicFileFormat.STATS_COLUMNS);
+ if (statsColumnsValue == null || statsColumnsValue.trim().isEmpty()) {
+ this.statsColumnNames = new ArrayList<>();
+ } else {
+ this.statsColumnNames =
+ Arrays.stream(statsColumnsValue.split(","))
+ .map(String::trim)
+ .filter(s -> !s.isEmpty())
+ .collect(Collectors.toList());
+ }
+ this.numBuckets = formatContext.options().get(MosaicFileFormat.NUM_BUCKETS);
+ }
+
+ @Override
+ public FormatWriter create(PositionOutputStream out, String compression) throws IOException {
Review Comment:
The argument from Paimon's writer path is currently ignored. always uses ' default compression (zstd) and only forwards the zstd level, so table options like or silently have no effect for Mosaic files. Please map supported Paimon compression values to Mosaic (or reject unsupported ones explicitly) and add coverage for at least and .
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "leaves12138 (via GitHub)" <gi...@apache.org>.
leaves12138 commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3278404020
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsWriter.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.arrow.ArrowBundleRecords;
+import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.BundleFormatWriter;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.io.BundleRecords;
+import org.apache.paimon.mosaic.ColumnStatistics;
+import org.apache.paimon.mosaic.MosaicWriter;
+import org.apache.paimon.mosaic.WriterOptions;
+import org.apache.paimon.types.RowType;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Mosaic records writer. */
+public class MosaicRecordsWriter implements BundleFormatWriter {
+
+ private final ArrowFormatWriter arrowFormatWriter;
+ private final MosaicWriter nativeWriter;
+ private final BufferAllocator allocator;
+ private final List<String> statsColumnNames;
+ @Nullable private MosaicWriterMetadata metadata;
+
+ public MosaicRecordsWriter(
+ OutputStream outputStream,
+ RowType rowType,
+ FileFormatFactory.FormatContext formatContext,
+ List<String> statsColumnNames,
+ int numBuckets,
+ String compression) {
+ this.statsColumnNames = statsColumnNames;
+ this.allocator = new RootAllocator();
+
+ int writeBatchSize = formatContext.writeBatchSize();
+ long writeBatchMemory = formatContext.writeBatchMemory().getBytes();
+
+ this.arrowFormatWriter =
+ new ArrowFormatWriter(rowType, writeBatchSize, true, allocator, writeBatchMemory);
+
+ Schema arrowSchema = arrowFormatWriter.getVectorSchemaRoot().getSchema();
+ WriterOptions options =
+ new WriterOptions()
+ .compression(resolveCompression(compression))
+ .zstdLevel(formatContext.zstdLevel())
+ .numBuckets(numBuckets)
+ .rowGroupMaxSize(writeBatchMemory);
Review Comment:
`rowGroupMaxSize` is currently derived from `write.batch-memory`. That option controls the Arrow writer buffer memory, while Paimon exposes `file.block-size` through `FormatContext.blockSize()` for format block / row-group sizing. As a result, Mosaic ignores `file.block-size`, and tuning `write.batch-memory` unexpectedly changes the Mosaic row-group layout. Please wire `formatContext.blockSize()` when it is configured, and otherwise leave Mosaic's own default (or document the intended default explicitly).
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicRecordsWriter.java:
##########
@@ -0,0 +1,211 @@
+/*
+ * 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.format.mosaic;
+
+import org.apache.paimon.arrow.ArrowBundleRecords;
+import org.apache.paimon.arrow.vector.ArrowFormatWriter;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.format.BundleFormatWriter;
+import org.apache.paimon.format.FileFormatFactory;
+import org.apache.paimon.io.BundleRecords;
+import org.apache.paimon.mosaic.ColumnStatistics;
+import org.apache.paimon.mosaic.MosaicWriter;
+import org.apache.paimon.mosaic.WriterOptions;
+import org.apache.paimon.types.RowType;
+
+import org.apache.arrow.memory.BufferAllocator;
+import org.apache.arrow.memory.RootAllocator;
+import org.apache.arrow.vector.VectorSchemaRoot;
+import org.apache.arrow.vector.types.pojo.Schema;
+
+import javax.annotation.Nullable;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** Mosaic records writer. */
+public class MosaicRecordsWriter implements BundleFormatWriter {
+
+ private final ArrowFormatWriter arrowFormatWriter;
+ private final MosaicWriter nativeWriter;
+ private final BufferAllocator allocator;
+ private final List<String> statsColumnNames;
+ @Nullable private MosaicWriterMetadata metadata;
+
+ public MosaicRecordsWriter(
+ OutputStream outputStream,
+ RowType rowType,
+ FileFormatFactory.FormatContext formatContext,
+ List<String> statsColumnNames,
+ int numBuckets,
+ String compression) {
+ this.statsColumnNames = statsColumnNames;
+ this.allocator = new RootAllocator();
+
+ int writeBatchSize = formatContext.writeBatchSize();
+ long writeBatchMemory = formatContext.writeBatchMemory().getBytes();
+
+ this.arrowFormatWriter =
+ new ArrowFormatWriter(rowType, writeBatchSize, true, allocator, writeBatchMemory);
+
+ Schema arrowSchema = arrowFormatWriter.getVectorSchemaRoot().getSchema();
+ WriterOptions options =
+ new WriterOptions()
+ .compression(resolveCompression(compression))
+ .zstdLevel(formatContext.zstdLevel())
+ .numBuckets(numBuckets)
+ .rowGroupMaxSize(writeBatchMemory);
+ if (!statsColumnNames.isEmpty()) {
+ options.statsColumns(statsColumnNames.toArray(new String[0]));
+ }
+
+ this.nativeWriter = new MosaicWriter(outputStream, arrowSchema, options, allocator);
+ }
+
+ private static int resolveCompression(String compression) {
+ if (compression == null) {
+ return WriterOptions.COMPRESSION_ZSTD;
+ }
+ switch (compression) {
+ case "none":
+ // return WriterOptions.COMPRESSION_NONE;
+ return 0;
+ case "zstd":
+ default:
Review Comment:
Thanks for wiring the compression argument. This still silently maps every value other than exact lower-case `none` and `zstd` to ZSTD. Paimon accepts values such as `lz4`, `snappy`, `gzip`, and options are generally case-insensitive; with `file.suffix.include.compression = true`, this can even create a `.lz4.mosaic` file whose Mosaic footer says ZSTD. Please normalize the value and fail fast for unsupported compressions instead of falling through to ZSTD, and add coverage for `none` plus an unsupported value. It would also be better for the Mosaic Java API to expose `COMPRESSION_NONE` instead of relying on magic `0`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [WIP][format] Add paimon-mosaic module with reader and writer [paimon]
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #7917:
URL: https://github.com/apache/paimon/pull/7917#discussion_r3286323925
##########
paimon-mosaic/src/main/java/org/apache/paimon/format/mosaic/MosaicInputFileAdapter.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.paimon.format.mosaic;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.fs.SeekableInputStream;
+import org.apache.paimon.mosaic.InputFile;
+
+import java.io.Closeable;
+import java.io.EOFException;
+import java.io.IOException;
+import java.util.ArrayDeque;
+import java.util.Deque;
+import java.util.concurrent.Semaphore;
+
+/**
+ * Adapts Paimon's {@link FileIO} to Mosaic's {@link InputFile} interface.
+ *
+ * <p>Maintains a pool of up to 8 {@link SeekableInputStream} instances. When all 8 are in use,
+ * callers block until one is returned. Thread-safe: Mosaic may invoke {@link #readFully}
+ * concurrently from multiple threads.
+ */
+public class MosaicInputFileAdapter implements InputFile, Closeable {
+
+ private static final int MAX_POOL_SIZE = 8;
+
+ private final FileIO fileIO;
+ private final Path path;
+ private final Semaphore semaphore;
+ private final Deque<SeekableInputStream> pool;
+ private boolean closed;
+
+ public MosaicInputFileAdapter(FileIO fileIO, Path path) {
+ this.fileIO = fileIO;
+ this.path = path;
+ this.semaphore = new Semaphore(MAX_POOL_SIZE);
+ this.pool = new ArrayDeque<>(MAX_POOL_SIZE);
+ }
+
+ @Override
+ public void readFully(long position, byte[] buffer, int offset, int length) throws IOException {
+ SeekableInputStream in = acquire();
Review Comment:
We should use pread.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@paimon.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org