Skip to content

Commit 1384d72

Browse files
jianguotianjianguotian
authored
[core][mosaic] Support MOSAIC in FormatTable (#8180)
1 parent 81a0375 commit 1384d72

3 files changed

Lines changed: 216 additions & 1 deletion

File tree

‎paimon-api/src/main/java/org/apache/paimon/CoreOptions.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -290,6 +290,7 @@ public InlineElement getDescription() {
290290
public static final String FILE_FORMAT_CSV = "csv";
291291
public static final String FILE_FORMAT_TEXT = "text";
292292
public static final String FILE_FORMAT_JSON = "json";
293+
public static final String FILE_FORMAT_MOSAIC = "mosaic";
293294

294295
public static final ConfigOption<String> FILE_FORMAT =
295296
key("file.format")
@@ -2902,6 +2903,7 @@ public String formatTableFileCompression() {
29022903
return "snappy";
29032904
case FILE_FORMAT_AVRO:
29042905
case FILE_FORMAT_ORC:
2906+
case FILE_FORMAT_MOSAIC:
29052907
return "zstd";
29062908
case FILE_FORMAT_CSV:
29072909
case FILE_FORMAT_TEXT:

‎paimon-core/src/main/java/org/apache/paimon/table/FormatTable.java‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,8 @@ enum Format {
7979
PARQUET,
8080
CSV,
8181
TEXT,
82-
JSON
82+
JSON,
83+
MOSAIC
8384
}
8485

8586
/** Parses a file format string to a corresponding {@link Format} enum constant. */

‎paimon-mosaic/src/test/java/org/apache/paimon/format/mosaic/FormatTableMosaicReadTest.java‎

Lines changed: 212 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,212 @@
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.paimon.format.mosaic;
20+
21+
import org.apache.paimon.catalog.Identifier;
22+
import org.apache.paimon.data.BinaryString;
23+
import org.apache.paimon.data.GenericRow;
24+
import org.apache.paimon.data.InternalRow;
25+
import org.apache.paimon.data.serializer.InternalRowSerializer;
26+
import org.apache.paimon.fs.Path;
27+
import org.apache.paimon.fs.local.LocalFileIO;
28+
import org.apache.paimon.reader.RecordReader;
29+
import org.apache.paimon.table.FormatTable;
30+
import org.apache.paimon.table.sink.BatchTableCommit;
31+
import org.apache.paimon.table.sink.BatchTableWrite;
32+
import org.apache.paimon.table.sink.BatchWriteBuilder;
33+
import org.apache.paimon.table.sink.CommitMessage;
34+
import org.apache.paimon.table.source.ReadBuilder;
35+
import org.apache.paimon.table.source.Split;
36+
import org.apache.paimon.types.DataTypes;
37+
import org.apache.paimon.types.RowType;
38+
39+
import org.junit.jupiter.api.BeforeAll;
40+
import org.junit.jupiter.api.Test;
41+
import org.junit.jupiter.api.io.TempDir;
42+
43+
import java.util.ArrayList;
44+
import java.util.Arrays;
45+
import java.util.Collections;
46+
import java.util.HashMap;
47+
import java.util.List;
48+
import java.util.Map;
49+
50+
import static org.assertj.core.api.Assertions.assertThat;
51+
import static org.assertj.core.api.Assertions.assertThatThrownBy;
52+
import static org.junit.jupiter.api.Assumptions.assumeTrue;
53+
54+
/** End-to-end tests for a {@link FormatTable} backed by mosaic files. */
55+
class FormatTableMosaicReadTest {
56+
57+
@TempDir java.nio.file.Path tempPath;
58+
59+
@BeforeAll
60+
static void checkNativeLibrary() {
61+
assumeTrue(isNativeAvailable(), "Mosaic native library not available");
62+
}
63+
64+
@Test
65+
void testReadWriteUnpartitioned() throws Exception {
66+
RowType rowType =
67+
RowType.builder()
68+
.field("a", DataTypes.STRING())
69+
.field("b", DataTypes.BIGINT())
70+
.field("c", DataTypes.DOUBLE())
71+
.build();
72+
FormatTable table = buildFormatTable(rowType, Collections.emptyList(), new HashMap<>());
73+
74+
writeAll(
75+
table,
76+
Arrays.asList(
77+
GenericRow.of(BinaryString.fromString("a1"), 1L, 1.1),
78+
GenericRow.of(BinaryString.fromString("a2"), 2L, 2.2),
79+
GenericRow.of(BinaryString.fromString("a3"), 3L, 3.3)));
80+
81+
List<InternalRow> result = readAll(table, rowType);
82+
83+
assertThat(result).hasSize(3);
84+
assertThat(result.get(0).getString(0).toString()).isEqualTo("a1");
85+
assertThat(result.get(0).getLong(1)).isEqualTo(1L);
86+
assertThat(result.get(0).getDouble(2)).isEqualTo(1.1);
87+
assertThat(result.get(2).getString(0).toString()).isEqualTo("a3");
88+
}
89+
90+
@Test
91+
void testReadWritePartitioned() throws Exception {
92+
RowType rowType =
93+
RowType.builder()
94+
.field("a", DataTypes.STRING())
95+
.field("b", DataTypes.BIGINT())
96+
.field("dt", DataTypes.STRING())
97+
.build();
98+
FormatTable table =
99+
buildFormatTable(rowType, Collections.singletonList("dt"), new HashMap<>());
100+
101+
writeAll(
102+
table,
103+
Arrays.asList(
104+
GenericRow.of(
105+
BinaryString.fromString("a1"),
106+
1L,
107+
BinaryString.fromString("20260608")),
108+
GenericRow.of(
109+
BinaryString.fromString("a2"),
110+
2L,
111+
BinaryString.fromString("20260608")),
112+
GenericRow.of(
113+
BinaryString.fromString("a3"),
114+
3L,
115+
BinaryString.fromString("20260609"))));
116+
117+
LocalFileIO fileIO = LocalFileIO.create();
118+
Path tablePath = new Path(table.location());
119+
assertThat(fileIO.exists(new Path(tablePath, "dt=20260608"))).isTrue();
120+
assertThat(fileIO.exists(new Path(tablePath, "dt=20260609"))).isTrue();
121+
122+
List<InternalRow> result = readAll(table, rowType);
123+
124+
assertThat(result).hasSize(3);
125+
List<String> aDt = new ArrayList<>();
126+
for (InternalRow row : result) {
127+
aDt.add(row.getString(0).toString() + "/" + row.getString(2).toString());
128+
}
129+
assertThat(aDt).containsExactlyInAnyOrder("a1/20260608", "a2/20260608", "a3/20260609");
130+
}
131+
132+
@Test
133+
void testEmptyDirectoryScansToNoSplits() throws Exception {
134+
RowType rowType = RowType.builder().field("a", DataTypes.STRING()).build();
135+
FormatTable table = buildFormatTable(rowType, Collections.emptyList(), new HashMap<>());
136+
137+
List<Split> splits = table.newReadBuilder().newScan().plan().splits();
138+
assertThat(splits).isEmpty();
139+
}
140+
141+
@Test
142+
void testInvalidCompressionPropagatedFromWriter() {
143+
RowType rowType = RowType.builder().field("a", DataTypes.STRING()).build();
144+
Map<String, String> options = new HashMap<>();
145+
options.put("file.compression", "gzip");
146+
FormatTable table = buildFormatTable(rowType, Collections.emptyList(), options);
147+
148+
assertThatThrownBy(
149+
() ->
150+
writeAll(
151+
table,
152+
Collections.singletonList(
153+
GenericRow.of(BinaryString.fromString("a1")))))
154+
.isInstanceOf(UnsupportedOperationException.class)
155+
.hasMessageContaining("Mosaic format only supports zstd");
156+
}
157+
158+
private FormatTable buildFormatTable(
159+
RowType rowType, List<String> partitionKeys, Map<String, String> options) {
160+
String location = new Path(tempPath.toUri()).toString();
161+
options.put("file.format", "mosaic");
162+
// CoreOptions.path() reads from options; SQL/REST catalogs inject it,
163+
// a raw Java-API caller has to do it explicitly.
164+
options.put("path", location);
165+
return FormatTable.builder()
166+
.fileIO(LocalFileIO.create())
167+
.identifier(Identifier.create("default", "t"))
168+
.rowType(rowType)
169+
.partitionKeys(partitionKeys)
170+
.location(location)
171+
.format(FormatTable.Format.MOSAIC)
172+
.options(options)
173+
.build();
174+
}
175+
176+
private static void writeAll(FormatTable table, List<InternalRow> rows) throws Exception {
177+
BatchWriteBuilder builder = table.newBatchWriteBuilder();
178+
try (BatchTableWrite write = builder.newWrite()) {
179+
for (InternalRow row : rows) {
180+
write.write(row);
181+
}
182+
List<CommitMessage> committables = write.prepareCommit();
183+
try (BatchTableCommit commit = builder.newCommit()) {
184+
commit.commit(committables);
185+
}
186+
}
187+
}
188+
189+
private static List<InternalRow> readAll(FormatTable table, RowType rowType) throws Exception {
190+
ReadBuilder readBuilder = table.newReadBuilder();
191+
List<Split> splits = readBuilder.newScan().plan().splits();
192+
assertThat(splits).isNotEmpty();
193+
194+
List<InternalRow> result = new ArrayList<>();
195+
InternalRowSerializer serializer = new InternalRowSerializer(rowType);
196+
for (Split split : splits) {
197+
try (RecordReader<InternalRow> reader = readBuilder.newRead().createReader(split)) {
198+
reader.forEachRemaining(row -> result.add(serializer.copy(row)));
199+
}
200+
}
201+
return result;
202+
}
203+
204+
private static boolean isNativeAvailable() {
205+
try {
206+
Class.forName("org.apache.paimon.mosaic.NativeLib");
207+
return true;
208+
} catch (Throwable t) {
209+
return false;
210+
}
211+
}
212+
}

0 commit comments

Comments
 (0)