You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@paimon.apache.org by "Alibaba-HZY (via GitHub)" <gi...@apache.org> on 2023/04/15 09:51:32 UTC
[GitHub] [incubator-paimon] Alibaba-HZY opened a new pull request, #918: [hive] Introduce Hive writer (#752)
Alibaba-HZY opened a new pull request, #918:
URL: https://github.com/apache/incubator-paimon/pull/918
Purpose
Introduce Hive writer,user can use hive sql to write data to the paimon table, for example :"insert into paimon_table select * from hive_table".Supports complex nested type writes such as: array<row<id bigint>>,map<string,array<decimal(5,2)>>,row<id bigint,name string>.Also fixed an issue where paimonRead could not read complex nested types.
Note that currently only mr is supported, tez is not supported, and only support single table writer.
https://github.com/apache/incubator-paimon/issues/752
Tests
TestPaimonStorageHandlerUseMR
API and Format
Documentation
Create a PaimonRecordWriter for each task in mapreduce, which contains a BatchTableWrite for writing data, It also contains a shared map that holds the mapping of task and PaimonRecordWriter, in order to obtain the precommit and commit data from BatchTableWrite in commitTask and commitJob, or abortCommit if the Job failed.
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172474361
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/utils/HiveUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hive.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.hive.PaimonJobConf;
+import org.apache.paimon.hive.SearchArgumentToPredicateConverter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+/** Utils for create {@link FileStoreTable} and {@link Predicate}. */
+public class HiveUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
+
+ public static FileStoreTable createFileStoreTable(JobConf jobConf) {
+ PaimonJobConf wrapper = new PaimonJobConf(jobConf);
+ Options options = PaimonJobConf.extractCatalogConfig(jobConf);
+ options.set(CoreOptions.PATH, wrapper.getLocation());
+ options.set(CoreOptions.WRITE_ONLY, Boolean.TRUE);
Review Comment:
add comments: why need to add this.
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1176449692
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class HiveWriteITCase {
+
+ @ClassRule public static TemporaryFolder folder = new TemporaryFolder();
+
+ @HiveSQL(files = {})
+ private static HiveShell hiveShell;
+
+ private static String engine;
+
+ private String commitUser;
+ private long commitIdentifier;
+
+ @BeforeClass
+ public static void beforeClass() {
+ // only run with mr
+ engine = "mr";
+ }
+
+ @Before
+ public void before() {
+ hiveShell.execute("SET hive.execution.engine=mr");
+
+ hiveShell.execute("CREATE DATABASE IF NOT EXISTS test_db");
+ hiveShell.execute("USE test_db");
+
+ commitUser = UUID.randomUUID().toString();
+ commitIdentifier = 0;
+ }
+
+ @After
+ public void after() {
+ hiveShell.execute("DROP DATABASE IF EXISTS test_db CASCADE");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data)
+ throws Exception {
+
+ return createChangelogExternalTable(rowType, partitionKeys, primaryKeys, data, "");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data,
+ String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, primaryKeys, identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data) throws Exception {
+ return createAppendOnlyExternalTable(rowType, partitionKeys, data, "");
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data, String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, Collections.emptyList(), identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String writeData(Table table, String path, List<InternalRow> data) throws Exception {
+ StreamWriteBuilder streamWriteBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ for (InternalRow rowData : data) {
+ write.write(rowData);
+ if (ThreadLocalRandom.current().nextInt(5) == 0) {
+ commit.commit(commitIdentifier, write.prepareCommit(false, commitIdentifier));
+ commitIdentifier++;
+ }
+ }
+ commit.commit(commitIdentifier, write.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
+ write.close();
+
+ String tableName = "test_table_" + (UUID.randomUUID().toString().substring(0, 4));
+ hiveShell.execute(
+ String.join(
+ "\n",
+ Arrays.asList(
+ "CREATE EXTERNAL TABLE " + tableName + " ",
+ "STORED BY '" + PaimonStorageHandler.class.getName() + "'",
+ "LOCATION '" + path + "'")));
+ return tableName;
+ }
+
+ @Test
+ public void testInsert() throws Exception {
+ List<InternalRow> emptyData = Collections.emptyList();
+
+ String outputTableName =
+ createAppendOnlyExternalTable(
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING()
+ },
+ new String[] {"pt", "a", "b", "c"}),
+ Collections.singletonList("pt"),
+ emptyData,
+ "hive_test_table_output");
+
+ hiveShell.execute(
Review Comment:
I have added a test for write-only option
--
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
[GitHub] [incubator-paimon] JingsongLi merged pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi merged PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1173239695
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/utils/HiveUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hive.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.hive.PaimonJobConf;
+import org.apache.paimon.hive.SearchArgumentToPredicateConverter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+/** Utils for create {@link FileStoreTable} and {@link Predicate}. */
+public class HiveUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
+
+ public static FileStoreTable createFileStoreTable(JobConf jobConf) {
+ PaimonJobConf wrapper = new PaimonJobConf(jobConf);
+ Options options = PaimonJobConf.extractCatalogConfig(jobConf);
+ options.set(CoreOptions.PATH, wrapper.getLocation());
+ options.set(CoreOptions.WRITE_ONLY, Boolean.TRUE);
Review Comment:
my bad~
I mean, if you add a test, please reply here which test was added
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174493595
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonSerDe.java:
##########
@@ -39,11 +42,18 @@
public class PaimonSerDe extends AbstractSerDe {
private PaimonInternalRowObjectInspector inspector;
+ private HiveSchema tableSchema;
+
+ private RowDataContainer rowData = new RowDataContainer();
+
+ private Map<ObjectInspector, HiveDeserializer> deserializers =
Review Comment:
Why not `final`?
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172346481
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/HiveDeserializer.java:
##########
@@ -0,0 +1,307 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalArray;
+import org.apache.paimon.data.InternalMap;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.objectinspector.HivePaimonArray;
+import org.apache.paimon.hive.objectinspector.WriteableObjectInspector;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.Preconditions;
+
+import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.StructField;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+/** A deserializer to deserialize hive objects to {@link InternalRow}. */
+public class HiveDeserializer {
+ private FieldDeserializer fieldDeserializer;
Review Comment:
final
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/SchemaVisitor.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/**
+ * SchemaVisitor to visitor the sourceInspector and writerInspector to get the FieldDeserializer.
+ */
+public abstract class SchemaVisitor<P, R> {
Review Comment:
replace this with `DataTypeDefaultVisitor` or `DataTypeVisitor`?
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputCommitter.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.hive.mapred;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.OutputCommitter;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** A Paimon table committer for adding data files to the Paimon table. */
+public class PaimonOutputCommitter extends OutputCommitter {
+
+ private static final String PRE_COMMIT = ".preCommit";
+
+ private static final Logger LOG = LoggerFactory.getLogger(PaimonOutputCommitter.class);
+
+ @Override
+ public void setupJob(JobContext jobContext) throws IOException {}
+
+ @Override
+ public void setupTask(TaskAttemptContext taskAttemptContext) throws IOException {}
+
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext taskAttemptContext) throws IOException {
+ // We need to commit if this is the last phase of a MapReduce process
+ return TaskType.REDUCE.equals(
+ taskAttemptContext.getTaskAttemptID().getTaskID().getTaskType())
+ || taskAttemptContext.getJobConf().getNumReduceTasks() == 0;
+ }
+
+ @Override
+ public void commitTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+ TaskAttemptID attemptID = taskAttemptContext.getTaskAttemptID();
+ JobConf jobConf = taskAttemptContext.getJobConf();
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ Map<String, PaimonRecordWriter> writers =
+ Optional.ofNullable(PaimonRecordWriter.getWriters(attemptID))
+ .orElseGet(
+ () -> {
+ LOG.info(
+ "CommitTask found no writers for output table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ return ImmutableMap.of();
+ });
+ PaimonRecordWriter writer = writers.get(table.name());
+ if (writer != null) {
+
+ try (BatchTableWrite batchTableWrite = writer.batchTableWrite()) {
+ List<CommitMessage> commitTables = batchTableWrite.prepareCommit();
+ createPreCommitFile(
+ commitTables,
+ generatePreCommitFileLocation(
+ table.location().getPath(),
+ attemptID.getJobID(),
+ attemptID.getTaskID().getId()),
+ table.fileIO());
+ } catch (Exception e) {
+ LOG.error(
+ "CommitTask prepareCommit error for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ throw new RuntimeException(e);
+ }
+ } else {
+ LOG.info(
+ "CommitTask found no writer for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ }
+ PaimonRecordWriter.removeWriters(attemptID);
+ }
+
+ @Override
+ public void abortTask(TaskAttemptContext taskAttemptContext) throws IOException {
+ Map<String, PaimonRecordWriter> writers =
+ PaimonRecordWriter.removeWriters(taskAttemptContext.getTaskAttemptID());
+
+ // close writer and delete files
+ if (writers != null) {
+ for (PaimonRecordWriter writer : writers.values()) {
+ writer.close(true);
+ }
+ }
+ }
+
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ JobConf jobConf = jobContext.getJobConf();
+
+ long startTime = System.currentTimeMillis();
+ LOG.info("CommitJob {} has started", jobContext.getJobID());
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ if (table != null) {
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
+ List<List<CommitMessage>> commitMessagesList =
+ getAllPreCommitMessage(table.location().getPath(), jobContext, table.fileIO());
+ for (List<CommitMessage> commitTables : commitMessagesList) {
+ try (BatchTableCommit batchTableCommit = batchWriteBuilder.newCommit()) {
+ batchTableCommit.commit(commitTables);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ deleteTemporaryFile(
+ jobContext,
+ generateJobLocation(table.location().getPath(), jobContext.getJobID()));
+ } else {
+ LOG.info("CommitJob not found table, Skipping job commit.");
+ }
+
+ LOG.info(
+ "Commit took {} ms for job {}",
+ System.currentTimeMillis() - startTime,
+ jobContext.getJobID());
+ }
+
+ @Override
+ public void abortJob(JobContext jobContext, int status) throws IOException {
+ FileStoreTable table = createFileStoreTable(jobContext.getJobConf());
+ if (table != null) {
+
+ LOG.info("AbortJob {} has started", jobContext.getJobID());
+ List<List<CommitMessage>> commitMessagesList =
+ getAllPreCommitMessage(table.location().getPath(), jobContext, table.fileIO());
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
+ for (List<CommitMessage> commitTables : commitMessagesList) {
+ try (BatchTableCommit batchTableCommit = batchWriteBuilder.newCommit()) {
+ batchTableCommit.abort(commitTables);
Review Comment:
Only one abort is OK
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/TestPaimonStorageHandlerUseMR.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class TestPaimonStorageHandlerUseMR {
Review Comment:
`HiveReadWriteITCase`?
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonSerDe.java:
##########
@@ -39,11 +42,18 @@
public class PaimonSerDe extends AbstractSerDe {
private PaimonInternalRowObjectInspector inspector;
+ private HiveSchema tableSchema;
+
+ private RowDataContainer rowData = new RowDataContainer();
+
+ private Map<ObjectInspector, HiveDeserializer> deserializers =
Review Comment:
final.
Note you have some fields without final.
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputCommitter.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.hive.mapred;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.OutputCommitter;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** A Paimon table committer for adding data files to the Paimon table. */
+public class PaimonOutputCommitter extends OutputCommitter {
+
+ private static final String PRE_COMMIT = ".preCommit";
+
+ private static final Logger LOG = LoggerFactory.getLogger(PaimonOutputCommitter.class);
+
+ @Override
+ public void setupJob(JobContext jobContext) throws IOException {}
+
+ @Override
+ public void setupTask(TaskAttemptContext taskAttemptContext) throws IOException {}
+
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext taskAttemptContext) throws IOException {
+ // We need to commit if this is the last phase of a MapReduce process
+ return TaskType.REDUCE.equals(
+ taskAttemptContext.getTaskAttemptID().getTaskID().getTaskType())
+ || taskAttemptContext.getJobConf().getNumReduceTasks() == 0;
+ }
+
+ @Override
+ public void commitTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+ TaskAttemptID attemptID = taskAttemptContext.getTaskAttemptID();
+ JobConf jobConf = taskAttemptContext.getJobConf();
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ Map<String, PaimonRecordWriter> writers =
+ Optional.ofNullable(PaimonRecordWriter.getWriters(attemptID))
+ .orElseGet(
+ () -> {
+ LOG.info(
+ "CommitTask found no writers for output table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ return ImmutableMap.of();
+ });
+ PaimonRecordWriter writer = writers.get(table.name());
+ if (writer != null) {
+
+ try (BatchTableWrite batchTableWrite = writer.batchTableWrite()) {
+ List<CommitMessage> commitTables = batchTableWrite.prepareCommit();
+ createPreCommitFile(
+ commitTables,
+ generatePreCommitFileLocation(
+ table.location().getPath(),
+ attemptID.getJobID(),
+ attemptID.getTaskID().getId()),
+ table.fileIO());
+ } catch (Exception e) {
+ LOG.error(
+ "CommitTask prepareCommit error for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ throw new RuntimeException(e);
+ }
+ } else {
+ LOG.info(
+ "CommitTask found no writer for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ }
+ PaimonRecordWriter.removeWriters(attemptID);
+ }
+
+ @Override
+ public void abortTask(TaskAttemptContext taskAttemptContext) throws IOException {
+ Map<String, PaimonRecordWriter> writers =
+ PaimonRecordWriter.removeWriters(taskAttemptContext.getTaskAttemptID());
+
+ // close writer and delete files
+ if (writers != null) {
+ for (PaimonRecordWriter writer : writers.values()) {
+ writer.close(true);
+ }
+ }
+ }
+
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ JobConf jobConf = jobContext.getJobConf();
+
+ long startTime = System.currentTimeMillis();
+ LOG.info("CommitJob {} has started", jobContext.getJobID());
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ if (table != null) {
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
+ List<List<CommitMessage>> commitMessagesList =
+ getAllPreCommitMessage(table.location().getPath(), jobContext, table.fileIO());
+ for (List<CommitMessage> commitTables : commitMessagesList) {
+ try (BatchTableCommit batchTableCommit = batchWriteBuilder.newCommit()) {
Review Comment:
Why there are multiple commits?
It is only a job? There should be just one commit.
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174501883
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputFormat.java:
##########
@@ -18,27 +18,70 @@
package org.apache.paimon.hive.mapred;
-import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.hive.RowDataContainer;
+import org.apache.paimon.shade.guava30.com.google.common.collect.Maps;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.TaskAttemptID;
import org.apache.hadoop.util.Progressable;
import java.io.IOException;
+import java.util.Map;
+import java.util.Properties;
-/** {@link OutputFormat} for table split. Currently useless. */
-public class PaimonOutputFormat implements OutputFormat<InternalRow, InternalRow> {
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** {@link OutputFormat} for table split. */
+public class PaimonOutputFormat
+ implements OutputFormat<NullWritable, RowDataContainer>,
+ HiveOutputFormat<NullWritable, RowDataContainer> {
+
+ private static final String TASK_ATTEMPT_ID_KEY = "mapreduce.task.attempt.id";
@Override
- public RecordWriter<InternalRow, InternalRow> getRecordWriter(
+ public RecordWriter<NullWritable, RowDataContainer> getRecordWriter(
FileSystem fileSystem, JobConf jobConf, String s, Progressable progressable)
throws IOException {
- throw new UnsupportedOperationException(
- "Paimon currently can only be used as an input format for Hive.");
+ return writer(jobConf);
}
@Override
public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException {}
+
+ @Override
+ public FileSinkOperator.RecordWriter getHiveRecordWriter(
+ JobConf jobConf,
+ Path path,
+ Class<? extends Writable> aClass,
+ boolean b,
+ Properties properties,
+ Progressable progressable)
+ throws IOException {
+ return writer(jobConf);
+ }
+
+ private static PaimonRecordWriter writer(JobConf jobConf) {
+ TaskAttemptID taskAttemptID = TaskAttemptID.forName(jobConf.get(TASK_ATTEMPT_ID_KEY));
+ FileStoreTable table = createFileStoreTable(jobConf);
+ Map<String, String> newOptions = Maps.newHashMapWithExpectedSize(1);
Review Comment:
Collections.singleMap
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172165633
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputFormat.java:
##########
@@ -18,27 +18,63 @@
package org.apache.paimon.hive.mapred;
-import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.RowDataContainer;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.TaskAttemptID;
import org.apache.hadoop.util.Progressable;
import java.io.IOException;
+import java.util.Properties;
-/** {@link OutputFormat} for table split. Currently useless. */
-public class PaimonOutputFormat implements OutputFormat<InternalRow, InternalRow> {
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** {@link OutputFormat} for table split. */
+public class PaimonOutputFormat
+ implements OutputFormat<NullWritable, RowDataContainer>,
+ HiveOutputFormat<NullWritable, RowDataContainer> {
+
+ private static final String TASK_ATTEMPT_ID_KEY = "mapreduce.task.attempt.id";
@Override
- public RecordWriter<InternalRow, InternalRow> getRecordWriter(
+ public RecordWriter<NullWritable, RowDataContainer> getRecordWriter(
FileSystem fileSystem, JobConf jobConf, String s, Progressable progressable)
throws IOException {
- throw new UnsupportedOperationException(
- "Paimon currently can only be used as an input format for Hive.");
+ return writer(jobConf);
}
@Override
public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException {}
+
+ @Override
+ public FileSinkOperator.RecordWriter getHiveRecordWriter(
+ JobConf jobConf,
+ Path path,
+ Class<? extends Writable> aClass,
+ boolean b,
+ Properties properties,
+ Progressable progressable)
+ throws IOException {
+ return writer(jobConf);
+ }
+
+ private static PaimonRecordWriter writer(JobConf jobConf) {
+ TaskAttemptID taskAttemptID = TaskAttemptID.forName(jobConf.get(TASK_ATTEMPT_ID_KEY));
+ FileStoreTable table = createFileStoreTable(jobConf);
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
Review Comment:
Here should we force write-only=true?
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174509761
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/DataTypeDefaultVisitor.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/**
+ * DataTypeDefaultVisitor to visitor the sourceInspector and writerInspector to get the
+ * FieldDeserializer.
+ */
+public abstract class DataTypeDefaultVisitor<P, R> {
Review Comment:
DataTypeDefaultVisitor or DataTypeVisitor may not apply in this case,they are have only one parameter,in hive we need tree param to deserialize value
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174493541
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/DataTypeDefaultVisitor.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/**
+ * DataTypeDefaultVisitor to visitor the sourceInspector and writerInspector to get the
+ * FieldDeserializer.
+ */
+public abstract class DataTypeDefaultVisitor<P, R> {
Review Comment:
I mean using `org.apache.paimon.types.DataTypeDefaultVisitor.`
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1176036329
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class HiveWriteITCase {
+
+ @ClassRule public static TemporaryFolder folder = new TemporaryFolder();
+
+ @HiveSQL(files = {})
+ private static HiveShell hiveShell;
+
+ private static String engine;
+
+ private String commitUser;
+ private long commitIdentifier;
+
+ @BeforeClass
+ public static void beforeClass() {
+ // only run with mr
+ engine = "mr";
+ }
+
+ @Before
+ public void before() {
+ hiveShell.execute("SET hive.execution.engine=mr");
+
+ hiveShell.execute("CREATE DATABASE IF NOT EXISTS test_db");
+ hiveShell.execute("USE test_db");
+
+ commitUser = UUID.randomUUID().toString();
+ commitIdentifier = 0;
+ }
+
+ @After
+ public void after() {
+ hiveShell.execute("DROP DATABASE IF EXISTS test_db CASCADE");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data)
+ throws Exception {
+
+ return createChangelogExternalTable(rowType, partitionKeys, primaryKeys, data, "");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data,
+ String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, primaryKeys, identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data) throws Exception {
+ return createAppendOnlyExternalTable(rowType, partitionKeys, data, "");
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data, String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, Collections.emptyList(), identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String writeData(Table table, String path, List<InternalRow> data) throws Exception {
+ StreamWriteBuilder streamWriteBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ for (InternalRow rowData : data) {
+ write.write(rowData);
+ if (ThreadLocalRandom.current().nextInt(5) == 0) {
+ commit.commit(commitIdentifier, write.prepareCommit(false, commitIdentifier));
+ commitIdentifier++;
+ }
+ }
+ commit.commit(commitIdentifier, write.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
+ write.close();
+
+ String tableName = "test_table_" + (UUID.randomUUID().toString().substring(0, 4));
+ hiveShell.execute(
+ String.join(
+ "\n",
+ Arrays.asList(
+ "CREATE EXTERNAL TABLE " + tableName + " ",
+ "STORED BY '" + PaimonStorageHandler.class.getName() + "'",
+ "LOCATION '" + path + "'")));
+ return tableName;
+ }
+
+ @Test
+ public void testInsert() throws Exception {
+ List<InternalRow> emptyData = Collections.emptyList();
+
+ String outputTableName =
+ createAppendOnlyExternalTable(
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING()
+ },
+ new String[] {"pt", "a", "b", "c"}),
+ Collections.singletonList("pt"),
+ emptyData,
+ "hive_test_table_output");
+
+ hiveShell.execute(
Review Comment:
If records in a bucket, 5 times are enough.
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172498894
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/TestPaimonStorageHandlerUseMR.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class TestPaimonStorageHandlerUseMR {
Review Comment:
HiveWriteITCase? This class is only for write test ,because only mr engine is supported
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1175238433
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class HiveWriteITCase {
+
+ @ClassRule public static TemporaryFolder folder = new TemporaryFolder();
+
+ @HiveSQL(files = {})
+ private static HiveShell hiveShell;
+
+ private static String engine;
+
+ private String commitUser;
+ private long commitIdentifier;
+
+ @BeforeClass
+ public static void beforeClass() {
+ // only run with mr
+ engine = "mr";
+ }
+
+ @Before
+ public void before() {
+ hiveShell.execute("SET hive.execution.engine=mr");
+
+ hiveShell.execute("CREATE DATABASE IF NOT EXISTS test_db");
+ hiveShell.execute("USE test_db");
+
+ commitUser = UUID.randomUUID().toString();
+ commitIdentifier = 0;
+ }
+
+ @After
+ public void after() {
+ hiveShell.execute("DROP DATABASE IF EXISTS test_db CASCADE");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data)
+ throws Exception {
+
+ return createChangelogExternalTable(rowType, partitionKeys, primaryKeys, data, "");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data,
+ String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, primaryKeys, identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data) throws Exception {
+ return createAppendOnlyExternalTable(rowType, partitionKeys, data, "");
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data, String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, Collections.emptyList(), identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String writeData(Table table, String path, List<InternalRow> data) throws Exception {
+ StreamWriteBuilder streamWriteBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ for (InternalRow rowData : data) {
+ write.write(rowData);
+ if (ThreadLocalRandom.current().nextInt(5) == 0) {
+ commit.commit(commitIdentifier, write.prepareCommit(false, commitIdentifier));
+ commitIdentifier++;
+ }
+ }
+ commit.commit(commitIdentifier, write.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
+ write.close();
+
+ String tableName = "test_table_" + (UUID.randomUUID().toString().substring(0, 4));
+ hiveShell.execute(
+ String.join(
+ "\n",
+ Arrays.asList(
+ "CREATE EXTERNAL TABLE " + tableName + " ",
+ "STORED BY '" + PaimonStorageHandler.class.getName() + "'",
+ "LOCATION '" + path + "'")));
+ return tableName;
+ }
+
+ @Test
+ public void testInsert() throws Exception {
+ List<InternalRow> emptyData = Collections.emptyList();
+
+ String outputTableName =
+ createAppendOnlyExternalTable(
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING()
+ },
+ new String[] {"pt", "a", "b", "c"}),
+ Collections.singletonList("pt"),
+ emptyData,
+ "hive_test_table_output");
+
+ hiveShell.execute(
Review Comment:
it may insert 50 times, because eacge file is too small to write 5 times without a compaction
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172193457
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputFormat.java:
##########
@@ -18,27 +18,63 @@
package org.apache.paimon.hive.mapred;
-import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.RowDataContainer;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.exec.FileSinkOperator;
+import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.Writable;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.TaskAttemptID;
import org.apache.hadoop.util.Progressable;
import java.io.IOException;
+import java.util.Properties;
-/** {@link OutputFormat} for table split. Currently useless. */
-public class PaimonOutputFormat implements OutputFormat<InternalRow, InternalRow> {
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** {@link OutputFormat} for table split. */
+public class PaimonOutputFormat
+ implements OutputFormat<NullWritable, RowDataContainer>,
+ HiveOutputFormat<NullWritable, RowDataContainer> {
+
+ private static final String TASK_ATTEMPT_ID_KEY = "mapreduce.task.attempt.id";
@Override
- public RecordWriter<InternalRow, InternalRow> getRecordWriter(
+ public RecordWriter<NullWritable, RowDataContainer> getRecordWriter(
FileSystem fileSystem, JobConf jobConf, String s, Progressable progressable)
throws IOException {
- throw new UnsupportedOperationException(
- "Paimon currently can only be used as an input format for Hive.");
+ return writer(jobConf);
}
@Override
public void checkOutputSpecs(FileSystem fileSystem, JobConf jobConf) throws IOException {}
+
+ @Override
+ public FileSinkOperator.RecordWriter getHiveRecordWriter(
+ JobConf jobConf,
+ Path path,
+ Class<? extends Writable> aClass,
+ boolean b,
+ Properties properties,
+ Progressable progressable)
+ throws IOException {
+ return writer(jobConf);
+ }
+
+ private static PaimonRecordWriter writer(JobConf jobConf) {
+ TaskAttemptID taskAttemptID = TaskAttemptID.forName(jobConf.get(TASK_ATTEMPT_ID_KEY));
+ FileStoreTable table = createFileStoreTable(jobConf);
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
Review Comment:
yes i have fixed
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172490372
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/mapred/PaimonOutputCommitter.java:
##########
@@ -0,0 +1,284 @@
+/*
+ * 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.hive.mapred;
+
+import org.apache.paimon.fs.FileIO;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.sink.BatchTableCommit;
+import org.apache.paimon.table.sink.BatchTableWrite;
+import org.apache.paimon.table.sink.BatchWriteBuilder;
+import org.apache.paimon.table.sink.CommitMessage;
+
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.JobContext;
+import org.apache.hadoop.mapred.OutputCommitter;
+import org.apache.hadoop.mapred.TaskAttemptContext;
+import org.apache.hadoop.mapred.TaskAttemptID;
+import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.paimon.hive.utils.HiveUtils.createFileStoreTable;
+
+/** A Paimon table committer for adding data files to the Paimon table. */
+public class PaimonOutputCommitter extends OutputCommitter {
+
+ private static final String PRE_COMMIT = ".preCommit";
+
+ private static final Logger LOG = LoggerFactory.getLogger(PaimonOutputCommitter.class);
+
+ @Override
+ public void setupJob(JobContext jobContext) throws IOException {}
+
+ @Override
+ public void setupTask(TaskAttemptContext taskAttemptContext) throws IOException {}
+
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext taskAttemptContext) throws IOException {
+ // We need to commit if this is the last phase of a MapReduce process
+ return TaskType.REDUCE.equals(
+ taskAttemptContext.getTaskAttemptID().getTaskID().getTaskType())
+ || taskAttemptContext.getJobConf().getNumReduceTasks() == 0;
+ }
+
+ @Override
+ public void commitTask(TaskAttemptContext taskAttemptContext) throws IOException {
+
+ TaskAttemptID attemptID = taskAttemptContext.getTaskAttemptID();
+ JobConf jobConf = taskAttemptContext.getJobConf();
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ Map<String, PaimonRecordWriter> writers =
+ Optional.ofNullable(PaimonRecordWriter.getWriters(attemptID))
+ .orElseGet(
+ () -> {
+ LOG.info(
+ "CommitTask found no writers for output table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ return ImmutableMap.of();
+ });
+ PaimonRecordWriter writer = writers.get(table.name());
+ if (writer != null) {
+
+ try (BatchTableWrite batchTableWrite = writer.batchTableWrite()) {
+ List<CommitMessage> commitTables = batchTableWrite.prepareCommit();
+ createPreCommitFile(
+ commitTables,
+ generatePreCommitFileLocation(
+ table.location().getPath(),
+ attemptID.getJobID(),
+ attemptID.getTaskID().getId()),
+ table.fileIO());
+ } catch (Exception e) {
+ LOG.error(
+ "CommitTask prepareCommit error for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ throw new RuntimeException(e);
+ }
+ } else {
+ LOG.info(
+ "CommitTask found no writer for specific table: {}, attemptID: {}",
+ table.name(),
+ attemptID);
+ }
+ PaimonRecordWriter.removeWriters(attemptID);
+ }
+
+ @Override
+ public void abortTask(TaskAttemptContext taskAttemptContext) throws IOException {
+ Map<String, PaimonRecordWriter> writers =
+ PaimonRecordWriter.removeWriters(taskAttemptContext.getTaskAttemptID());
+
+ // close writer and delete files
+ if (writers != null) {
+ for (PaimonRecordWriter writer : writers.values()) {
+ writer.close(true);
+ }
+ }
+ }
+
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ JobConf jobConf = jobContext.getJobConf();
+
+ long startTime = System.currentTimeMillis();
+ LOG.info("CommitJob {} has started", jobContext.getJobID());
+ FileStoreTable table = createFileStoreTable(jobConf);
+
+ if (table != null) {
+ BatchWriteBuilder batchWriteBuilder = table.newBatchWriteBuilder();
+ List<List<CommitMessage>> commitMessagesList =
+ getAllPreCommitMessage(table.location().getPath(), jobContext, table.fileIO());
+ for (List<CommitMessage> commitTables : commitMessagesList) {
+ try (BatchTableCommit batchTableCommit = batchWriteBuilder.newCommit()) {
Review Comment:
yes We only need one commit
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172474361
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/utils/HiveUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hive.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.hive.PaimonJobConf;
+import org.apache.paimon.hive.SearchArgumentToPredicateConverter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+/** Utils for create {@link FileStoreTable} and {@link Predicate}. */
+public class HiveUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
+
+ public static FileStoreTable createFileStoreTable(JobConf jobConf) {
+ PaimonJobConf wrapper = new PaimonJobConf(jobConf);
+ Options options = PaimonJobConf.extractCatalogConfig(jobConf);
+ options.set(CoreOptions.PATH, wrapper.getLocation());
+ options.set(CoreOptions.WRITE_ONLY, Boolean.TRUE);
Review Comment:
We need to add tests for this. You add wrong place. We can should use `FileStoreTable.copy` to add write-only, but here is catalog options.
Please reply this thread to the new test case.
--
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
[GitHub] [incubator-paimon] JingsongLi commented on pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#issuecomment-1510652144
Wow~ Thanks @Alibaba-HZY !
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174500381
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/PaimonSerDe.java:
##########
@@ -39,11 +42,18 @@
public class PaimonSerDe extends AbstractSerDe {
private PaimonInternalRowObjectInspector inspector;
+ private HiveSchema tableSchema;
+
+ private RowDataContainer rowData = new RowDataContainer();
+
+ private Map<ObjectInspector, HiveDeserializer> deserializers =
Review Comment:
ok my bad ,deserializers is also can be final
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174501676
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/HiveWriteITCase.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class HiveWriteITCase {
+
+ @ClassRule public static TemporaryFolder folder = new TemporaryFolder();
+
+ @HiveSQL(files = {})
+ private static HiveShell hiveShell;
+
+ private static String engine;
+
+ private String commitUser;
+ private long commitIdentifier;
+
+ @BeforeClass
+ public static void beforeClass() {
+ // only run with mr
+ engine = "mr";
+ }
+
+ @Before
+ public void before() {
+ hiveShell.execute("SET hive.execution.engine=mr");
+
+ hiveShell.execute("CREATE DATABASE IF NOT EXISTS test_db");
+ hiveShell.execute("USE test_db");
+
+ commitUser = UUID.randomUUID().toString();
+ commitIdentifier = 0;
+ }
+
+ @After
+ public void after() {
+ hiveShell.execute("DROP DATABASE IF EXISTS test_db CASCADE");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data)
+ throws Exception {
+
+ return createChangelogExternalTable(rowType, partitionKeys, primaryKeys, data, "");
+ }
+
+ private String createChangelogExternalTable(
+ RowType rowType,
+ List<String> partitionKeys,
+ List<String> primaryKeys,
+ List<InternalRow> data,
+ String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, primaryKeys, identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data) throws Exception {
+ return createAppendOnlyExternalTable(rowType, partitionKeys, data, "");
+ }
+
+ private String createAppendOnlyExternalTable(
+ RowType rowType, List<String> partitionKeys, List<InternalRow> data, String tableName)
+ throws Exception {
+ String path = folder.newFolder().toURI().toString();
+ String tableNameNotNull =
+ StringUtils.isNullOrWhitespaceOnly(tableName) ? TABLE_NAME : tableName;
+ String tablePath = String.format("%s/default.db/%s", path, tableNameNotNull);
+ Options conf = new Options();
+ conf.set(CatalogOptions.WAREHOUSE, path);
+ conf.set(CoreOptions.BUCKET, 2);
+ conf.set(CoreOptions.FILE_FORMAT, CoreOptions.FileFormatType.AVRO);
+ conf.set(CoreOptions.WRITE_MODE, WriteMode.APPEND_ONLY);
+ Identifier identifier = Identifier.create(DATABASE_NAME, tableNameNotNull);
+ Table table =
+ FileStoreTestUtils.createFileStoreTable(
+ conf, rowType, partitionKeys, Collections.emptyList(), identifier);
+
+ return writeData(table, tablePath, data);
+ }
+
+ private String writeData(Table table, String path, List<InternalRow> data) throws Exception {
+ StreamWriteBuilder streamWriteBuilder = table.newStreamWriteBuilder();
+ StreamTableWrite write = streamWriteBuilder.newWrite();
+ StreamTableCommit commit = streamWriteBuilder.newCommit();
+ for (InternalRow rowData : data) {
+ write.write(rowData);
+ if (ThreadLocalRandom.current().nextInt(5) == 0) {
+ commit.commit(commitIdentifier, write.prepareCommit(false, commitIdentifier));
+ commitIdentifier++;
+ }
+ }
+ commit.commit(commitIdentifier, write.prepareCommit(true, commitIdentifier));
+ commitIdentifier++;
+ write.close();
+
+ String tableName = "test_table_" + (UUID.randomUUID().toString().substring(0, 4));
+ hiveShell.execute(
+ String.join(
+ "\n",
+ Arrays.asList(
+ "CREATE EXTERNAL TABLE " + tableName + " ",
+ "STORED BY '" + PaimonStorageHandler.class.getName() + "'",
+ "LOCATION '" + path + "'")));
+ return tableName;
+ }
+
+ @Test
+ public void testInsert() throws Exception {
+ List<InternalRow> emptyData = Collections.emptyList();
+
+ String outputTableName =
+ createAppendOnlyExternalTable(
+ RowType.of(
+ new DataType[] {
+ DataTypes.INT(),
+ DataTypes.INT(),
+ DataTypes.BIGINT(),
+ DataTypes.STRING()
+ },
+ new String[] {"pt", "a", "b", "c"}),
+ Collections.singletonList("pt"),
+ emptyData,
+ "hive_test_table_output");
+
+ hiveShell.execute(
Review Comment:
We can add a case: insert 5 times, and see there is no compaction. (See no compact snapshots)
To validate `write-only` option.
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174308968
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/utils/HiveUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hive.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.hive.PaimonJobConf;
+import org.apache.paimon.hive.SearchArgumentToPredicateConverter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+/** Utils for create {@link FileStoreTable} and {@link Predicate}. */
+public class HiveUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
+
+ public static FileStoreTable createFileStoreTable(JobConf jobConf) {
+ PaimonJobConf wrapper = new PaimonJobConf(jobConf);
+ Options options = PaimonJobConf.extractCatalogConfig(jobConf);
+ options.set(CoreOptions.PATH, wrapper.getLocation());
+ options.set(CoreOptions.WRITE_ONLY, Boolean.TRUE);
Review Comment:
> my bad~ I mean, if you add a test, please reply here which test was added
ok i removed this option in here,i use FileStoreTable.copy to add write-only
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1174514070
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/DataTypeDefaultVisitor.java:
##########
@@ -0,0 +1,117 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.shade.guava30.com.google.common.collect.Lists;
+import org.apache.paimon.types.ArrayType;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.MapType;
+import org.apache.paimon.types.RowType;
+
+import java.util.List;
+
+/**
+ * DataTypeDefaultVisitor to visitor the sourceInspector and writerInspector to get the
+ * FieldDeserializer.
+ */
+public abstract class DataTypeDefaultVisitor<P, R> {
Review Comment:
Maybe return a custom class to contains tree param?
--
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
[GitHub] [incubator-paimon] Alibaba-HZY commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "Alibaba-HZY (via GitHub)" <gi...@apache.org>.
Alibaba-HZY commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1172538595
##########
paimon-hive/paimon-hive-connector-common/src/main/java/org/apache/paimon/hive/utils/HiveUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * 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.hive.utils;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.catalog.CatalogContext;
+import org.apache.paimon.hive.PaimonJobConf;
+import org.apache.paimon.hive.SearchArgumentToPredicateConverter;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.schema.TableSchema;
+import org.apache.paimon.table.FileStoreTable;
+import org.apache.paimon.table.FileStoreTableFactory;
+
+import org.apache.hadoop.hive.ql.io.sarg.ConvertAstToSearchArg;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.mapred.JobConf;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+/** Utils for create {@link FileStoreTable} and {@link Predicate}. */
+public class HiveUtils {
+
+ private static final Logger LOG = LoggerFactory.getLogger(HiveUtils.class);
+
+ public static FileStoreTable createFileStoreTable(JobConf jobConf) {
+ PaimonJobConf wrapper = new PaimonJobConf(jobConf);
+ Options options = PaimonJobConf.extractCatalogConfig(jobConf);
+ options.set(CoreOptions.PATH, wrapper.getLocation());
+ options.set(CoreOptions.WRITE_ONLY, Boolean.TRUE);
Review Comment:
ok i will use FileStoreTable.copy to add write-only,but i do not know what that means of 'reply this thread to the new test case.'
--
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
[GitHub] [incubator-paimon] JingsongLi commented on a diff in pull request #918: [hive] Introduce Hive writer (#752)
Posted by "JingsongLi (via GitHub)" <gi...@apache.org>.
JingsongLi commented on code in PR #918:
URL: https://github.com/apache/incubator-paimon/pull/918#discussion_r1173239808
##########
paimon-hive/paimon-hive-connector-common/src/test/java/org/apache/paimon/hive/TestPaimonStorageHandlerUseMR.java:
##########
@@ -0,0 +1,1033 @@
+/*
+ * 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.hive;
+
+import org.apache.paimon.CoreOptions;
+import org.apache.paimon.WriteMode;
+import org.apache.paimon.catalog.Identifier;
+import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.Decimal;
+import org.apache.paimon.data.GenericArray;
+import org.apache.paimon.data.GenericMap;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.hive.mapred.PaimonOutputFormat;
+import org.apache.paimon.hive.runner.PaimonEmbeddedHiveRunner;
+import org.apache.paimon.options.CatalogOptions;
+import org.apache.paimon.options.Options;
+import org.apache.paimon.table.Table;
+import org.apache.paimon.table.sink.StreamTableCommit;
+import org.apache.paimon.table.sink.StreamTableWrite;
+import org.apache.paimon.table.sink.StreamWriteBuilder;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataType;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowKind;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.StringUtils;
+
+import com.klarna.hiverunner.HiveShell;
+import com.klarna.hiverunner.annotations.HiveSQL;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.junit.runner.RunWith;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ThreadLocalRandom;
+
+import static org.apache.paimon.hive.FileStoreTestUtils.DATABASE_NAME;
+import static org.apache.paimon.hive.FileStoreTestUtils.TABLE_NAME;
+import static org.apache.paimon.hive.RandomGenericRowDataGenerator.randomBigDecimal;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** IT cases for {@link PaimonStorageHandler} and {@link PaimonOutputFormat}. */
+@RunWith(PaimonEmbeddedHiveRunner.class)
+public class TestPaimonStorageHandlerUseMR {
Review Comment:
OK
--
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