You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2022/08/01 02:21:18 UTC
[flink] 02/02: [FLINK-26366][hive] Support INSERT OVERWRITE DIRECTORY statement
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a42c0041fc37595d07010c2077b2f806b2b34ef9
Author: luoyuxia <lu...@alibaba-inc.com>
AuthorDate: Fri Apr 15 17:54:11 2022 +0800
[FLINK-26366][hive] Support INSERT OVERWRITE DIRECTORY statement
This closes #19520
---
.../connector/file/table/FileSystemCommitter.java | 15 ++-
.../file/table/FileSystemOutputFormat.java | 17 +++-
.../connector/file/table/FileSystemTableSink.java | 4 +-
.../connector/file/table/PartitionLoader.java | 34 +++++--
.../flink/connectors/hive/HiveTableSink.java | 113 +++++++++++++++++----
.../table/catalog/hive/util/HiveTableUtil.java | 11 +-
.../delegation/hive/HiveParserConstants.java | 3 +
.../delegation/hive/HiveParserDMLHelper.java | 93 ++++++++++++++++-
.../hive/copy/HiveParserDirectoryDesc.java | 43 ++++++++
.../planner/delegation/hive/copy/HiveParserQB.java | 14 ++-
.../hive/copy/HiveParserSemanticAnalyzer.java | 35 ++-----
.../hive/parse/HiveParserDDLSemanticAnalyzer.java | 2 +-
.../connectors/hive/HiveDialectQueryITCase.java | 56 ++++++++++
13 files changed, 370 insertions(+), 70 deletions(-)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
index 2cb8a909038..69ee3c0695b 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemCommitter.java
@@ -55,6 +55,7 @@ class FileSystemCommitter implements Serializable {
private final FileSystemFactory factory;
private final TableMetaStoreFactory metaStoreFactory;
private final boolean overwrite;
+ private final boolean isToLocal;
private final Path tmpPath;
private final int partitionColumnSize;
@@ -64,11 +65,22 @@ class FileSystemCommitter implements Serializable {
boolean overwrite,
Path tmpPath,
int partitionColumnSize) {
+ this(factory, metaStoreFactory, overwrite, tmpPath, partitionColumnSize, false);
+ }
+
+ FileSystemCommitter(
+ FileSystemFactory factory,
+ TableMetaStoreFactory metaStoreFactory,
+ boolean overwrite,
+ Path tmpPath,
+ int partitionColumnSize,
+ boolean isToLocal) {
this.factory = factory;
this.metaStoreFactory = metaStoreFactory;
this.overwrite = overwrite;
this.tmpPath = tmpPath;
this.partitionColumnSize = partitionColumnSize;
+ this.isToLocal = isToLocal;
}
/** For committing job's output after successful batch job completion. */
@@ -76,7 +88,8 @@ class FileSystemCommitter implements Serializable {
FileSystem fs = factory.create(tmpPath.toUri());
List<Path> taskPaths = listTaskTemporaryPaths(fs, tmpPath);
- try (PartitionLoader loader = new PartitionLoader(overwrite, fs, metaStoreFactory)) {
+ try (PartitionLoader loader =
+ new PartitionLoader(overwrite, fs, metaStoreFactory, isToLocal)) {
if (partitionColumnSize > 0) {
for (Map.Entry<LinkedHashMap<String, String>, List<Path>> entry :
collectPartSpecToPaths(fs, taskPaths, partitionColumnSize).entrySet()) {
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
index 5eb03f1651c..006a7b16713 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemOutputFormat.java
@@ -46,6 +46,7 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
private final FileSystemFactory fsFactory;
private final TableMetaStoreFactory msFactory;
private final boolean overwrite;
+ private final boolean isToLocal;
private final Path tmpPath;
private final String[] partitionColumns;
private final boolean dynamicGrouped;
@@ -61,6 +62,7 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
FileSystemFactory fsFactory,
TableMetaStoreFactory msFactory,
boolean overwrite,
+ boolean isToLocal,
Path tmpPath,
String[] partitionColumns,
boolean dynamicGrouped,
@@ -71,6 +73,7 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
this.fsFactory = fsFactory;
this.msFactory = msFactory;
this.overwrite = overwrite;
+ this.isToLocal = isToLocal;
this.tmpPath = tmpPath;
this.partitionColumns = partitionColumns;
this.dynamicGrouped = dynamicGrouped;
@@ -85,7 +88,12 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
try {
FileSystemCommitter committer =
new FileSystemCommitter(
- fsFactory, msFactory, overwrite, tmpPath, partitionColumns.length);
+ fsFactory,
+ msFactory,
+ overwrite,
+ tmpPath,
+ partitionColumns.length,
+ isToLocal);
committer.commitPartitions();
} catch (Exception e) {
throw new TableException("Exception in finalizeGlobal", e);
@@ -149,6 +157,7 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
private LinkedHashMap<String, String> staticPartitions = new LinkedHashMap<>();
private boolean dynamicGrouped = false;
private boolean overwrite = false;
+ private boolean isToLocal = false;
private FileSystemFactory fileSystemFactory = FileSystem::get;
private PartitionComputer<T> computer;
@@ -190,6 +199,11 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
return this;
}
+ public Builder<T> setIsToLocal(boolean isToLocal) {
+ this.isToLocal = isToLocal;
+ return this;
+ }
+
public Builder<T> setTempPath(Path tmpPath) {
this.tmpPath = tmpPath;
return this;
@@ -216,6 +230,7 @@ public class FileSystemOutputFormat<T> implements OutputFormat<T>, FinalizeOnMas
fileSystemFactory,
metaStoreFactory,
overwrite,
+ isToLocal,
tmpPath,
partitionColumns,
dynamicGrouped,
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java
index 81eb49a5457..b7e24c37d86 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSink.java
@@ -181,9 +181,7 @@ public class FileSystemTableSink extends AbstractFileSystemTable
builder.setStaticPartitions(staticPartitions);
builder.setTempPath(toStagingPath());
builder.setOutputFileConfig(
- OutputFileConfig.builder()
- .withPartPrefix("part-" + UUID.randomUUID().toString())
- .build());
+ OutputFileConfig.builder().withPartPrefix("part-" + UUID.randomUUID()).build());
return inputStream
.writeUsingOutputFormat(builder.build())
.setParallelism(parallelism)
diff --git a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
index f8be5e5ef8d..a85ec543a63 100644
--- a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
+++ b/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/PartitionLoader.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.Internal;
import org.apache.flink.core.fs.FileStatus;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
+import org.apache.flink.util.FileUtils;
import org.apache.flink.util.Preconditions;
import java.io.Closeable;
@@ -49,12 +50,19 @@ public class PartitionLoader implements Closeable {
private final boolean overwrite;
private final FileSystem fs;
private final TableMetaStoreFactory.TableMetaStore metaStore;
+ // whether it's to load files to local file system
+ private final boolean isToLocal;
- public PartitionLoader(boolean overwrite, FileSystem fs, TableMetaStoreFactory factory)
+ public PartitionLoader(
+ boolean overwrite,
+ FileSystem sourceFs,
+ TableMetaStoreFactory factory,
+ boolean isToLocal)
throws Exception {
this.overwrite = overwrite;
- this.fs = fs;
+ this.fs = sourceFs;
this.metaStore = factory.createTableMetaStore();
+ this.isToLocal = isToLocal;
}
/** Load a single partition. */
@@ -68,27 +76,28 @@ public class PartitionLoader implements Closeable {
metaStore.getLocationPath(),
generatePartitionPath(partSpec)));
- overwriteAndRenameFiles(srcDirs, path);
+ overwriteAndMoveFiles(srcDirs, path);
metaStore.createOrAlterPartition(partSpec, path);
}
/** Load a non-partition files to output path. */
public void loadNonPartition(List<Path> srcDirs) throws Exception {
Path tableLocation = metaStore.getLocationPath();
- overwriteAndRenameFiles(srcDirs, tableLocation);
+ overwriteAndMoveFiles(srcDirs, tableLocation);
}
- private void overwriteAndRenameFiles(List<Path> srcDirs, Path destDir) throws Exception {
- boolean dirSuccessExist = fs.exists(destDir) || fs.mkdirs(destDir);
+ private void overwriteAndMoveFiles(List<Path> srcDirs, Path destDir) throws Exception {
+ FileSystem destFileSystem = destDir.getFileSystem();
+ boolean dirSuccessExist = destFileSystem.exists(destDir) || destFileSystem.mkdirs(destDir);
Preconditions.checkState(dirSuccessExist, "Failed to create dest path " + destDir);
overwrite(destDir);
- renameFiles(srcDirs, destDir);
+ moveFiles(srcDirs, destDir);
}
private void overwrite(Path destDir) throws Exception {
if (overwrite) {
// delete existing files for overwrite
- FileStatus[] existingFiles = listStatusWithoutHidden(fs, destDir);
+ FileStatus[] existingFiles = listStatusWithoutHidden(destDir.getFileSystem(), destDir);
if (existingFiles != null) {
for (FileStatus existingFile : existingFiles) {
// TODO: We need move to trash when auto-purge is false.
@@ -99,7 +108,7 @@ public class PartitionLoader implements Closeable {
}
/** Moves files from srcDir to destDir. */
- private void renameFiles(List<Path> srcDirs, Path destDir) throws Exception {
+ private void moveFiles(List<Path> srcDirs, Path destDir) throws Exception {
for (Path srcDir : srcDirs) {
if (!srcDir.equals(destDir)) {
FileStatus[] srcFiles = listStatusWithoutHidden(fs, srcDir);
@@ -107,7 +116,12 @@ public class PartitionLoader implements Closeable {
for (FileStatus srcFile : srcFiles) {
Path srcPath = srcFile.getPath();
Path destPath = new Path(destDir, srcPath.getName());
- fs.rename(srcPath, destPath);
+ // if it's not to move to local file system, just rename it
+ if (!isToLocal) {
+ fs.rename(srcPath, destPath);
+ } else {
+ FileUtils.copy(srcPath, destPath, true);
+ }
}
}
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
index abbc2786994..fc9dfb84828 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/connectors/hive/HiveTableSink.java
@@ -22,10 +22,12 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.connector.file.table.EmptyMetaStoreFactory;
import org.apache.flink.connector.file.table.FileSystemConnectorOptions;
import org.apache.flink.connector.file.table.FileSystemOutputFormat;
import org.apache.flink.connector.file.table.FileSystemTableSink;
import org.apache.flink.connector.file.table.FileSystemTableSink.TableBucketAssigner;
+import org.apache.flink.connector.file.table.TableMetaStoreFactory;
import org.apache.flink.connector.file.table.stream.PartitionCommitInfo;
import org.apache.flink.connector.file.table.stream.StreamingSink;
import org.apache.flink.connector.file.table.stream.compact.CompactReader;
@@ -47,6 +49,7 @@ import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSin
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink.BucketsBuilder;
import org.apache.flink.streaming.api.functions.sink.filesystem.rollingpolicies.CheckpointRollingPolicy;
import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
import org.apache.flink.table.catalog.CatalogTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
import org.apache.flink.table.catalog.exceptions.CatalogException;
@@ -56,6 +59,7 @@ import org.apache.flink.table.catalog.hive.client.HiveShim;
import org.apache.flink.table.catalog.hive.client.HiveShimLoader;
import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
import org.apache.flink.table.catalog.hive.util.HiveReflectionUtils;
+import org.apache.flink.table.catalog.hive.util.HiveTableUtil;
import org.apache.flink.table.connector.ChangelogMode;
import org.apache.flink.table.connector.ProviderContext;
import org.apache.flink.table.connector.sink.DataStreamSinkProvider;
@@ -102,6 +106,8 @@ import static org.apache.flink.connector.file.table.FileSystemConnectorOptions.S
import static org.apache.flink.connector.file.table.FileSystemConnectorOptions.SINK_ROLLING_POLICY_ROLLOVER_INTERVAL;
import static org.apache.flink.connector.file.table.stream.compact.CompactOperator.convertToUncompacted;
import static org.apache.flink.table.catalog.hive.util.HiveTableUtil.checkAcidTable;
+import static org.apache.flink.table.planner.delegation.hive.HiveParserConstants.IS_INSERT_DIRECTORY;
+import static org.apache.flink.table.planner.delegation.hive.HiveParserConstants.IS_TO_LOCAL_DIRECTORY;
/** Table sink to write to Hive tables. */
public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, SupportsOverwrite {
@@ -179,11 +185,46 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
DataStructureConverter converter) {
checkAcidTable(catalogTable.getOptions(), identifier.toObjectPath());
- try (HiveMetastoreClientWrapper client =
- HiveMetastoreClientFactory.create(HiveConfUtils.create(jobConf), hiveVersion)) {
- Table table = client.getTable(identifier.getDatabaseName(), identifier.getObjectName());
- StorageDescriptor sd = table.getSd();
+ StorageDescriptor sd;
+ Properties tableProps = new Properties();
+ Map<String, String> options = catalogTable.getOptions();
+ boolean isInsertDirectory =
+ Boolean.parseBoolean(
+ options.getOrDefault(
+ CatalogPropertiesUtil.FLINK_PROPERTY_PREFIX + IS_INSERT_DIRECTORY,
+ "false"));
+ boolean isToLocal = false;
+ if (isInsertDirectory) {
+ isToLocal =
+ Boolean.parseBoolean(
+ options.getOrDefault(
+ CatalogPropertiesUtil.FLINK_PROPERTY_PREFIX
+ + IS_TO_LOCAL_DIRECTORY,
+ "false"));
+ sd =
+ org.apache.hadoop.hive.ql.metadata.Table.getEmptyTable(
+ identifier.getDatabaseName(), identifier.getObjectName())
+ .getSd();
+ HiveConf hiveConf = HiveConfUtils.create(jobConf);
+ HiveTableUtil.setDefaultStorageFormatForDirectory(sd, HiveConfUtils.create(jobConf));
+ HiveTableUtil.extractRowFormat(sd, catalogTable.getOptions());
+ HiveTableUtil.extractStoredAs(sd, catalogTable.getOptions(), hiveConf);
+ HiveTableUtil.extractLocation(sd, catalogTable.getOptions());
+ tableProps.putAll(sd.getSerdeInfo().getParameters());
+ tableProps.putAll(catalogTable.getOptions());
+ } else {
+ try (HiveMetastoreClientWrapper client =
+ HiveMetastoreClientFactory.create(HiveConfUtils.create(jobConf), hiveVersion)) {
+ Table table =
+ client.getTable(identifier.getDatabaseName(), identifier.getObjectName());
+ sd = table.getSd();
+ tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, table);
+ } catch (TException e) {
+ throw new CatalogException("Failed to query Hive metaStore", e);
+ }
+ }
+ try {
Class hiveOutputFormatClz =
hiveShim.getHiveOutputFormatClass(Class.forName(sd.getOutputFormat()));
boolean isCompressed =
@@ -195,7 +236,7 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
sd.getSerdeInfo(),
tableSchema,
getPartitionKeyArray(),
- HiveReflectionUtils.getTableMetadata(hiveShim, table),
+ tableProps,
hiveShim,
isCompressed);
String extension =
@@ -206,21 +247,49 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
OutputFileConfig.OutputFileConfigBuilder fileNamingBuilder =
OutputFileConfig.builder()
- .withPartPrefix("part-" + UUID.randomUUID().toString())
+ .withPartPrefix("part-" + UUID.randomUUID())
.withPartSuffix(extension == null ? "" : extension);
final int parallelism =
Optional.ofNullable(configuredParallelism).orElse(dataStream.getParallelism());
if (isBounded) {
OutputFileConfig fileNaming = fileNamingBuilder.build();
+ TableMetaStoreFactory msFactory =
+ isInsertDirectory
+ ? new EmptyMetaStoreFactory(
+ new org.apache.flink.core.fs.Path(sd.getLocation()))
+ : msFactory();
+ // default to use the dest location as the parent directory of staging directory
+ String stagingParentDir = sd.getLocation();
+ if (isToLocal) {
+ // it's for writing to local file system, dest location is a path in local file
+ // system. we need to know the scratch path for
+ // non-local file system, so that it'll first write to the scratch path and then
+ // move to the local path
+ Path rootScratchDirPath =
+ new Path(
+ HiveConf.getVar(
+ HiveConfUtils.create(jobConf),
+ HiveConf.ConfVars.SCRATCHDIR));
+ // TODO: may append something more meaningful than a timestamp, like query ID
+ Path scratchDir =
+ new Path(
+ rootScratchDirPath, String.valueOf(System.currentTimeMillis()));
+ stagingParentDir = scratchDir.toUri().toString();
+ }
return createBatchSink(
- dataStream, converter, sd, writerFactory, fileNaming, parallelism);
+ dataStream,
+ converter,
+ writerFactory,
+ msFactory,
+ fileNaming,
+ stagingParentDir,
+ isToLocal,
+ parallelism);
} else {
if (overwrite) {
throw new IllegalStateException("Streaming mode not support overwrite.");
}
-
- Properties tableProps = HiveReflectionUtils.getTableMetadata(hiveShim, table);
return createStreamSink(
providerContext,
dataStream,
@@ -230,8 +299,6 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
fileNamingBuilder,
parallelism);
}
- } catch (TException e) {
- throw new CatalogException("Failed to query Hive metaStore", e);
} catch (IOException e) {
throw new FlinkRuntimeException("Failed to create staging dir", e);
} catch (ClassNotFoundException e) {
@@ -244,9 +311,11 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
private DataStreamSink<Row> createBatchSink(
DataStream<RowData> dataStream,
DataStructureConverter converter,
- StorageDescriptor sd,
HiveWriterFactory recordWriterFactory,
+ TableMetaStoreFactory metaStoreFactory,
OutputFileConfig fileNaming,
+ String stagingParentDir,
+ boolean isToLocal,
final int parallelism)
throws IOException {
FileSystemOutputFormat.Builder<Row> builder = new FileSystemOutputFormat.Builder<>();
@@ -261,11 +330,12 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
builder.setPartitionColumns(getPartitionKeyArray());
builder.setFileSystemFactory(fsFactory());
builder.setFormatFactory(new HiveOutputFormatFactory(recordWriterFactory));
- builder.setMetaStoreFactory(msFactory());
+ builder.setMetaStoreFactory(metaStoreFactory);
builder.setOverwrite(overwrite);
+ builder.setIsToLocal(isToLocal);
builder.setStaticPartitions(staticPartitionSpec);
builder.setTempPath(
- new org.apache.flink.core.fs.Path(toStagingDir(sd.getLocation(), jobConf)));
+ new org.apache.flink.core.fs.Path(toStagingDir(stagingParentDir, jobConf)));
builder.setOutputFileConfig(fileNaming);
return dataStream
.map((MapFunction<RowData, Row>) value -> (Row) converter.toExternal(value))
@@ -460,20 +530,19 @@ public class HiveTableSink implements DynamicTableSink, SupportsPartitioning, Su
return supportsGrouping;
}
- // get a staging dir associated with a final dir
- private String toStagingDir(String finalDir, Configuration conf) throws IOException {
- String res = finalDir;
- if (!finalDir.endsWith(Path.SEPARATOR)) {
- res += Path.SEPARATOR;
+ // get a staging dir
+ private String toStagingDir(String stagingParentDir, Configuration conf) throws IOException {
+ if (!stagingParentDir.endsWith(Path.SEPARATOR)) {
+ stagingParentDir += Path.SEPARATOR;
}
// TODO: may append something more meaningful than a timestamp, like query ID
- res += ".staging_" + System.currentTimeMillis();
- Path path = new Path(res);
+ stagingParentDir += ".staging_" + System.currentTimeMillis();
+ Path path = new Path(stagingParentDir);
FileSystem fs = path.getFileSystem(conf);
Preconditions.checkState(
fs.exists(path) || fs.mkdirs(path), "Failed to create staging dir " + path);
fs.deleteOnExit(path);
- return res;
+ return stagingParentDir;
}
private List<String> getPartitionKeys() {
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
index 08c108c0b44..7db7d9fad1b 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/catalog/hive/util/HiveTableUtil.java
@@ -60,6 +60,7 @@ import org.apache.hadoop.hive.ql.io.RCFileStorageFormatDescriptor;
import org.apache.hadoop.hive.ql.io.StorageFormatDescriptor;
import org.apache.hadoop.hive.ql.io.StorageFormatFactory;
import org.apache.hadoop.hive.serde.serdeConstants;
+import org.apache.hadoop.hive.serde2.lazy.LazySimpleSerDe;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import java.io.File;
@@ -298,7 +299,7 @@ public class HiveTableUtil {
}
}
- private static void extractStoredAs(
+ public static void extractStoredAs(
StorageDescriptor sd, Map<String, String> properties, HiveConf hiveConf) {
String storageFormat = properties.remove(STORED_AS_FILE_FORMAT);
String inputFormat = properties.remove(STORED_AS_INPUT_FORMAT);
@@ -333,6 +334,14 @@ public class HiveTableUtil {
setStorageFormat(sd, hiveConf.getVar(HiveConf.ConfVars.HIVEDEFAULTFILEFORMAT), hiveConf);
}
+ public static void setDefaultStorageFormatForDirectory(
+ StorageDescriptor sd, HiveConf hiveConf) {
+ // default is LazySimpleSerDe for insert into directory
+ sd.getSerdeInfo().setSerializationLib(LazySimpleSerDe.class.getName());
+ // default is TextFile
+ setStorageFormat(sd, "TextFile", hiveConf);
+ }
+
public static void alterColumns(StorageDescriptor sd, CatalogTable catalogTable) {
List<FieldSchema> allCols = HiveTableUtil.createHiveColumns(catalogTable.getSchema());
List<FieldSchema> nonPartCols =
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserConstants.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserConstants.java
index ee2d3f7b362..3bd3e0ff469 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserConstants.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserConstants.java
@@ -27,4 +27,7 @@ public class HiveParserConstants {
/* Constants for Druid storage handler */
public static final String DRUID_HIVE_STORAGE_HANDLER_ID =
"org.apache.hadoop.hive.druid.DruidStorageHandler";
+ /* Constants for insert overwrite directory */
+ public static final String IS_INSERT_DIRECTORY = "is-insert-directory";
+ public static final String IS_TO_LOCAL_DIRECTORY = "is-to-local-directory";
}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserDMLHelper.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserDMLHelper.java
index eacc8d8341c..17e207c505f 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserDMLHelper.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/HiveParserDMLHelper.java
@@ -19,17 +19,27 @@
package org.apache.flink.table.planner.delegation.hive;
import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.table.api.Schema;
import org.apache.flink.table.catalog.CatalogManager;
+import org.apache.flink.table.catalog.CatalogPropertiesUtil;
+import org.apache.flink.table.catalog.CatalogTable;
+import org.apache.flink.table.catalog.ContextResolvedTable;
import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.catalog.ResolvedCatalogTable;
+import org.apache.flink.table.catalog.ResolvedSchema;
import org.apache.flink.table.catalog.UnresolvedIdentifier;
import org.apache.flink.table.catalog.hive.HiveCatalog;
+import org.apache.flink.table.catalog.hive.factories.HiveCatalogFactoryOptions;
+import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.operations.Operation;
import org.apache.flink.table.operations.QueryOperation;
import org.apache.flink.table.operations.SinkModifyOperation;
import org.apache.flink.table.planner.delegation.PlannerContext;
+import org.apache.flink.table.planner.delegation.hive.copy.HiveParserDirectoryDesc;
import org.apache.flink.table.planner.delegation.hive.copy.HiveParserQB;
import org.apache.flink.table.planner.delegation.hive.copy.HiveParserSqlFunctionConverter;
import org.apache.flink.table.planner.delegation.hive.copy.HiveParserTypeConverter;
+import org.apache.flink.table.planner.delegation.hive.parse.HiveParserDDLSemanticAnalyzer;
import org.apache.flink.table.planner.operations.PlannerQueryOperation;
import org.apache.flink.table.planner.plan.nodes.hive.LogicalDistribution;
import org.apache.flink.table.planner.plan.nodes.hive.LogicalScriptTransform;
@@ -61,11 +71,13 @@ import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.SettableUDF;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoFactory;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
+import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
@@ -73,6 +85,10 @@ import java.util.Set;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
+import static org.apache.flink.sql.parser.hive.ddl.SqlCreateHiveTable.TABLE_LOCATION_URI;
+import static org.apache.flink.table.planner.delegation.hive.HiveParserConstants.IS_INSERT_DIRECTORY;
+import static org.apache.flink.table.planner.delegation.hive.HiveParserConstants.IS_TO_LOCAL_DIRECTORY;
+
/** A helper class to handle DMLs in hive dialect. */
public class HiveParserDMLHelper {
@@ -227,7 +243,7 @@ public class HiveParserDMLHelper {
// decide the dest table
Map<String, Table> nameToDestTable = qbMetaData.getNameToDestTable();
Map<String, Partition> nameToDestPart = qbMetaData.getNameToDestPartition();
- // for now we only support inserting to a single table
+ // for now we only support inserting to a single table in one queryRelNode
Preconditions.checkState(
nameToDestTable.size() <= 1 && nameToDestPart.size() <= 1,
"Only support inserting to 1 table");
@@ -241,7 +257,7 @@ public class HiveParserDMLHelper {
destTable = nameToDestPart.values().iterator().next().getTable();
} else {
// happens for INSERT DIRECTORY
- throw new SemanticException("INSERT DIRECTORY is not supported");
+ return createInsertIntoDirectoryOperation(topQB, qbMetaData, queryRelNode);
}
// decide static partition specs
@@ -296,6 +312,79 @@ public class HiveParserDMLHelper {
Collections.emptyMap());
}
+ private SinkModifyOperation createInsertIntoDirectoryOperation(
+ HiveParserQB topQB, QBMetaData qbMetaData, RelNode queryRelNode) {
+ String dest = topQB.getParseInfo().getClauseNamesForDest().iterator().next();
+ // get the location for insert into directory
+ String location = qbMetaData.getDestFileForAlias(dest);
+ // get whether it's for insert local directory
+ boolean isToLocal = qbMetaData.getDestTypeForAlias(dest) == QBMetaData.DEST_LOCAL_FILE;
+ HiveParserDirectoryDesc directoryDesc = topQB.getDirectoryDesc();
+
+ // set row format / stored as / location
+ Map<String, String> props = new HashMap<>();
+ HiveParserDDLSemanticAnalyzer.encodeRowFormat(directoryDesc.getRowFormatParams(), props);
+ HiveParserDDLSemanticAnalyzer.encodeStorageFormat(directoryDesc.getStorageFormat(), props);
+ props.put(TABLE_LOCATION_URI, location);
+
+ props.put(FactoryUtil.CONNECTOR.key(), HiveCatalogFactoryOptions.IDENTIFIER);
+ // mark it's for insert into directory
+ props.put(CatalogPropertiesUtil.FLINK_PROPERTY_PREFIX + IS_INSERT_DIRECTORY, "true");
+ // mark it's for insert into local directory or not
+ props.put(
+ CatalogPropertiesUtil.FLINK_PROPERTY_PREFIX + IS_TO_LOCAL_DIRECTORY,
+ String.valueOf(isToLocal));
+
+ List<RelDataTypeField> fieldList = queryRelNode.getRowType().getFieldList();
+ String[] colNameArr = new String[fieldList.size()];
+ String[] colTypeArr = new String[fieldList.size()];
+ for (int i = 0; i < fieldList.size(); i++) {
+ colNameArr[i] = fieldList.get(i).getName();
+ TypeInfo typeInfo = HiveParserTypeConverter.convert(fieldList.get(i).getType());
+ if (typeInfo.equals(TypeInfoFactory.voidTypeInfo)) {
+ colTypeArr[i] = TypeInfoFactory.stringTypeInfo.getTypeName();
+ } else {
+ colTypeArr[i] = typeInfo.getTypeName();
+ }
+ }
+
+ // extra information needed for insert into directory
+ String colNames = String.join(",", colNameArr);
+ String colTypes = String.join(":", colTypeArr);
+ props.put("columns", colNames);
+ props.put("columns.types", colTypes);
+
+ PlannerQueryOperation plannerQueryOperation = new PlannerQueryOperation(queryRelNode);
+ return new SinkModifyOperation(
+ createDummyTableForInsertDirectory(
+ plannerQueryOperation.getResolvedSchema(), props),
+ plannerQueryOperation,
+ Collections.emptyMap(),
+ true, // insert into directory is always for overwrite
+ Collections.emptyMap());
+ }
+
+ private ContextResolvedTable createDummyTableForInsertDirectory(
+ ResolvedSchema resolvedSchema, Map<String, String> props) {
+ Schema schema = Schema.newBuilder().fromResolvedSchema(resolvedSchema).build();
+ CatalogTable catalogTable =
+ CatalogTable.of(
+ schema,
+ "a dummy table for the case of insert overwrite directory ",
+ Collections.emptyList(),
+ props);
+ ResolvedCatalogTable resolvedCatalogTable =
+ new ResolvedCatalogTable(catalogTable, resolvedSchema);
+ String currentCatalog = catalogManager.getCurrentCatalog();
+ // the object name means nothing, it's just for placeholder and won't be used actually
+ String objectName = "insert_directory_tbl";
+ return ContextResolvedTable.permanent(
+ ObjectIdentifier.of(
+ currentCatalog, catalogManager.getCurrentDatabase(), objectName),
+ catalogManager.getCatalog(currentCatalog).get(),
+ resolvedCatalogTable);
+ }
+
private RelNode replaceDistForStaticParts(
LogicalDistribution hiveDist,
Table destTable,
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserDirectoryDesc.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserDirectoryDesc.java
new file mode 100644
index 00000000000..53be2e131be
--- /dev/null
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserDirectoryDesc.java
@@ -0,0 +1,43 @@
+/*
+ * 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.flink.table.planner.delegation.hive.copy;
+
+/**
+ * Counterpart of hive's org.apache.hadoop.hive.ql.plan.CreateTableDesc, but only contains the
+ * properties needed for "insert overwrite directory".
+ */
+public class HiveParserDirectoryDesc {
+ private final HiveParserBaseSemanticAnalyzer.HiveParserRowFormatParams rowFormatParams;
+ private final HiveParserStorageFormat storageFormat;
+
+ public HiveParserDirectoryDesc(
+ HiveParserBaseSemanticAnalyzer.HiveParserRowFormatParams rowFormatParams,
+ HiveParserStorageFormat storageFormat) {
+ this.rowFormatParams = rowFormatParams;
+ this.storageFormat = storageFormat;
+ }
+
+ public HiveParserBaseSemanticAnalyzer.HiveParserRowFormatParams getRowFormatParams() {
+ return rowFormatParams;
+ }
+
+ public HiveParserStorageFormat getStorageFormat() {
+ return storageFormat;
+ }
+}
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserQB.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserQB.java
index f8f98c4e671..bdccd09b7f0 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserQB.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserQB.java
@@ -69,6 +69,10 @@ public class HiveParserQB {
*/
private int numSubQueryPredicates;
+ private CreateTableDesc createTableDesc;
+
+ private HiveParserDirectoryDesc directoryDesc;
+
public void print(String msg) {
LOG.info(msg + "alias=" + qbp.getAlias());
for (String alias : getSubqAliases()) {
@@ -219,10 +223,16 @@ public class HiveParserQB {
}
public CreateTableDesc getTableDesc() {
- return null;
+ return createTableDesc;
}
- public void setDirectoryDesc(CreateTableDesc directoryDesc) {}
+ public void setDirectoryDesc(HiveParserDirectoryDesc directoryDesc) {
+ this.directoryDesc = directoryDesc;
+ }
+
+ public HiveParserDirectoryDesc getDirectoryDesc() {
+ return directoryDesc;
+ }
public boolean isCTAS() {
return false;
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSemanticAnalyzer.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSemanticAnalyzer.java
index 591421dde7e..5275493c860 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSemanticAnalyzer.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/copy/HiveParserSemanticAnalyzer.java
@@ -74,7 +74,6 @@ import org.apache.hadoop.hive.ql.parse.PTFInvocationSpec.PTFQueryInputType;
import org.apache.hadoop.hive.ql.parse.PrunedPartitionList;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.parse.SplitSample;
-import org.apache.hadoop.hive.ql.plan.CreateTableDesc;
import org.apache.hadoop.hive.ql.plan.CreateViewDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeColumnDesc;
import org.apache.hadoop.hive.ql.plan.ExprNodeDesc;
@@ -1793,37 +1792,25 @@ public class HiveParserSemanticAnalyzer {
}
}
- boolean isDfsFile = true;
- if (ast.getChildCount() >= 2
- && ast.getChild(1).getText().toLowerCase().equals("local")) {
- isDfsFile = false;
- }
+ boolean isDfsFile =
+ ast.getChildCount() < 2
+ || !ast.getChild(1).getText().equalsIgnoreCase("local");
// Set the destination for the SELECT query inside the CTAS
qb.getMetaData().setDestForAlias(name, fname, isDfsFile);
- CreateTableDesc directoryDesc = new CreateTableDesc();
- boolean directoryDescIsSet = false;
+ // we use a dedicated class to represent for 'insert overwrite directory'
+ HiveParserDirectoryDesc directoryDesc =
+ new HiveParserDirectoryDesc(rowFormatParams, storageFormat);
int numCh = ast.getChildCount();
for (int num = 1; num < numCh; num++) {
HiveParserASTNode child = (HiveParserASTNode) ast.getChild(num);
if (child != null) {
if (storageFormat.fillStorageFormat(child)) {
- directoryDesc.setOutputFormat(storageFormat.getOutputFormat());
- directoryDesc.setSerName(storageFormat.getSerde());
- directoryDescIsSet = true;
continue;
}
switch (child.getToken().getType()) {
case HiveASTParser.TOK_TABLEROWFORMAT:
rowFormatParams.analyzeRowFormat(child);
- directoryDesc.setFieldDelim(rowFormatParams.fieldDelim);
- directoryDesc.setLineDelim(rowFormatParams.lineDelim);
- directoryDesc.setCollItemDelim(
- rowFormatParams.collItemDelim);
- directoryDesc.setMapKeyDelim(rowFormatParams.mapKeyDelim);
- directoryDesc.setFieldEscape(rowFormatParams.fieldEscape);
- directoryDesc.setNullFormat(rowFormatParams.nullFormat);
- directoryDescIsSet = true;
break;
case HiveASTParser.TOK_TABLESERIALIZER:
HiveParserASTNode serdeChild =
@@ -1831,23 +1818,17 @@ public class HiveParserSemanticAnalyzer {
storageFormat.setSerde(
unescapeSQLString(
serdeChild.getChild(0).getText()));
- directoryDesc.setSerName(storageFormat.getSerde());
if (serdeChild.getChildCount() > 1) {
- directoryDesc.setSerdeProps(
- new HashMap<String, String>());
readProps(
(HiveParserASTNode)
serdeChild.getChild(1).getChild(0),
- directoryDesc.getSerdeProps());
+ storageFormat.getSerdeProps());
}
- directoryDescIsSet = true;
break;
}
}
}
- if (directoryDescIsSet) {
- qb.setDirectoryDesc(directoryDesc);
- }
+ qb.setDirectoryDesc(directoryDesc);
break;
}
default:
diff --git a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
index e1f13801fd3..9874da8faf0 100644
--- a/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
+++ b/flink-connectors/flink-connector-hive/src/main/java/org/apache/flink/table/planner/delegation/hive/parse/HiveParserDDLSemanticAnalyzer.java
@@ -1180,7 +1180,7 @@ public class HiveParserDDLSemanticAnalyzer {
}
}
- private void encodeStorageFormat(
+ public static void encodeStorageFormat(
HiveParserStorageFormat storageFormat, Map<String, String> props) {
String serdeName = storageFormat.getSerde();
if (serdeName != null) {
diff --git a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java
index 8e79f6dad93..0314325b2bd 100644
--- a/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java
+++ b/flink-connectors/flink-connector-hive/src/test/java/org/apache/flink/connectors/hive/HiveDialectQueryITCase.java
@@ -28,6 +28,7 @@ import org.apache.flink.table.module.hive.HiveModule;
import org.apache.flink.table.planner.delegation.hive.HiveParserUtils;
import org.apache.flink.types.Row;
import org.apache.flink.util.CollectionUtil;
+import org.apache.flink.util.FileUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.ql.exec.UDFArgumentException;
@@ -44,6 +45,8 @@ import org.junit.Test;
import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -436,6 +439,59 @@ public class HiveDialectQueryITCase {
.hasMessage(expectedMessage);
}
+ @Test
+ public void testInsertDirectory() throws Exception {
+ String warehouse = hiveCatalog.getHiveConf().getVar(HiveConf.ConfVars.METASTOREWAREHOUSE);
+
+ // test insert overwrite directory with row format parameters
+ tableEnv.executeSql("create table map_table (foo STRING , bar MAP<STRING, INT>)");
+ tableEnv.executeSql(
+ "insert into map_table select 'A', map('math',100,'english',90,'history',85)");
+
+ String dataDir = warehouse + "/map_table_dir";
+ tableEnv.executeSql(
+ String.format(
+ "INSERT OVERWRITE LOCAL DIRECTORY '%s'"
+ + "ROW FORMAT DELIMITED \n"
+ + "FIELDS TERMINATED BY ':'\n"
+ + "COLLECTION ITEMS TERMINATED BY '#' \n"
+ + "MAP KEYS TERMINATED BY '=' select * from map_table",
+ dataDir))
+ .await();
+ java.nio.file.Path[] files =
+ FileUtils.listFilesInDirectory(
+ Paths.get(dataDir), (path) -> !path.toFile().isHidden())
+ .toArray(new Path[0]);
+ assertThat(files.length).isEqualTo(1);
+ String actualString = FileUtils.readFileUtf8(files[0].toFile());
+ assertThat(actualString.trim()).isEqualTo("A:english=90#math=100#history=85");
+
+ // test insert overwrite directory store as other format
+ tableEnv.executeSql("create table d_table(x int) PARTITIONED BY (ds STRING, hr STRING)");
+ tableEnv.executeSql("INSERT OVERWRITE TABLE d_table PARTITION (ds='1', hr='1') select 1")
+ .await();
+ tableEnv.executeSql("INSERT OVERWRITE TABLE d_table PARTITION (ds='1', hr='2') select 2")
+ .await();
+
+ String tableAggDir = warehouse + "/d_table_agg";
+ // create an external referring to the directory to be inserted
+ tableEnv.executeSql(
+ String.format(
+ "create external table d_table_agg(x int, ds STRING) STORED AS RCFILE location '%s' ",
+ tableAggDir));
+ // insert into directory stored as RCFILE
+ tableEnv.executeSql(
+ String.format(
+ "INSERT OVERWRITE DIRECTORY '%s' STORED AS RCFILE select count(x), ds from d_table group by ds ",
+ tableAggDir))
+ .await();
+ List<Row> result =
+ CollectionUtil.iteratorToList(
+ tableEnv.executeSql("select * from d_table_agg").collect());
+ // verify the data read from the external table
+ assertThat(result.toString()).isEqualTo("[+I[2, 1]]");
+ }
+
@Test
public void testScriptTransform() throws Exception {
tableEnv.executeSql("CREATE TABLE dest1(key INT, ten INT, one INT, value STRING)");