<commit_before>/*
 * 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.connectors.hive;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.description.Description;
import org.apache.flink.configuration.description.InlineElement;
import org.apache.flink.connector.file.table.FileSystemConnectorOptions;

import java.time.Duration;

import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.configuration.description.TextElement.text;

/** This class holds configuration constants used by hive connector. */
public class HiveOptions {

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER =
            key("table.exec.hive.fallback-mapred-reader")
                    .booleanType()
                    .defaultValue(false)
                    .withDescription(
                            "If it is false, using flink native vectorized reader to read orc files; "
                                    + "If it is true, using hadoop mapred record reader to read orc files.");

    public static final ConfigOption<Boolean>
            TABLE_EXEC_HIVE_READ_PARTITION_WITH_SUBDIRECTORY_ENABLED =
                    key("table.exec.hive.read-partition-with-subdirectory.enabled")
                            .booleanType()
                            .defaultValue(true)
                            .withDescription(
                                    "If it is true, flink will read the files of partitioned hive table from subdirectories under the partition directory to be read.\n"
                                            + "If it is false, an exception that 'not a file: xxx' will be thrown when the partition directory contains any sub-directory.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM =
            key("table.exec.hive.infer-source-parallelism")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If is false, parallelism of source are set by config.\n"
                                    + "If is true, source parallelism is inferred according to splits number.\n");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX =
            key("table.exec.hive.infer-source-parallelism.max")
                    .intType()
                    .defaultValue(1000)
                    .withDescription("Sets max infer parallelism for source operator.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_WRITER =
            key("table.exec.hive.fallback-mapred-writer")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If it is false, using flink native writer to write parquet and orc files; "
                                    + "If it is true, using hadoop mapred record writer to write parquet and orc files.");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM =
            key("table.exec.hive.load-partition-splits.thread-num")
                    .intType()
                    .defaultValue(3)
                    .withDescription(
                            "The thread number to split hive's partitions to splits. It should be bigger than 0.");

    public static final ConfigOption<MemorySize> TABLE_EXEC_HIVE_SPLIT_MAX_BYTES =
            key("table.exec.hive.split-max-size")
                    .memoryType()
                    .defaultValue(MemorySize.parse("128mb"))
                    .withDescription(
                            "The maximum number of bytes (default is 128MB) to pack into a split while reading Hive table. A split will be assigned to a reader. "
                                    + "It only works for the Hive table stored as ORC format.");

    public static final ConfigOption<MemorySize> TABLE_EXEC_HIVE_FILE_OPEN_COST =
            key("table.exec.hive.file-open-cost")
                    .memoryType()
                    .defaultValue(MemorySize.parse("4mb"))
                    .withDescription(
                            "The estimated cost (default is 4MB) to open a file. Used to split Hive's files to splits."
                                    + " When the value is over estimated, Flink will tend to pack Hive's data into less splits, which will be helpful when Hive's table contains many small files."
                                    + " And vice versa. It only works for the Hive table stored as ORC format.");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_CALCULATE_PARTITION_SIZE_THREAD_NUM =
            key("table.exec.hive.calculate-partition-size.thread-num")
                    .intType()
                    .defaultValue(3)
                    .withDeprecatedKeys("The thread number to calculate partition's size.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_DYNAMIC_GROUPING_ENABLED =
            key("table.exec.hive.sink.sort-by-dynamic-partition.enable")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "Whether to enable sorting data by dynamic partition column or not when it's for dynamic partition writing."
                                    + " The default is to enable sorting."
                                    + " If it's enabled, the date will be sorted additionally by the dynamic partition columns"
                                    + " before writing into sink table."
                                    + " If it's disabled, there won't be extra sorting,"
                                    + " but it may throw OutOfMemory exception if there are too many dynamic partitions fall into same sink node."
                                    + " Note: it only works in batch mode.");

    /**
     * Hive users usually commit partition for metastore and a _SUCCESS file. That's why we create a
     * same option with {@link FileSystemConnectorOptions#SINK_PARTITION_COMMIT_POLICY_KIND} with
     * different default value.
     */
    public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_KIND =
            key("sink.partition-commit.policy.kind")
                    .stringType()
                    .defaultValue("metastore,success-file")
                    .withDescription(
                            "Policy to commit a partition is to notify the downstream"
                                    + " application that the partition has finished writing, the partition"
                                    + " is ready to be read."
                                    + " metastore: add partition to metastore. "
                                    + " success-file: add a success file to the partition directory. The success file name can be configured by the 'sink.partition-commit.success-file.name' option."
                                    + " Both can be configured at the same time: 'metastore,success-file'."
                                    + " custom: use policy class to create a commit policy."
                                    + " Support to configure multiple policies: 'metastore,success-file'.");

    public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_CLASS =
            FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS;

    public static final ConfigOption<String> SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME =
            FileSystemConnectorOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME;

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_SINK_STATISTIC_AUTO_GATHER_ENABLE =
            key("table.exec.hive.sink.statistic-auto-gather.enable")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If it's true, Flink will gather statistic automatically during writing Hive Table."
                                    + " If it's false, no any statistic will be gathered. The default value is true."
                                    + " For ORC and Parquet format, numFiles/totalSize/numRows/rawDataSize can be gathered."
                                    + " For other format, only numFiles/totalSize can be gathered."
                                    + " Note: only batch mode supports auto gather statistic, stream mode doesn't support it yet.");

    public static final ConfigOption<Integer>
            TABLE_EXEC_HIVE_SINK_STATISTIC_AUTO_GATHER_THREAD_NUM =
                    key("table.exec.hive.sink.statistic-auto-gather.thread-num")
                            .intType()
                            .defaultValue(3)
                            .withDeprecatedKeys(
                                    "The number of threads used to gather statistic during writing Hive Table"
                                            + " when the table is stored as ORC or Parquet format."
                                            + " The default value is 3.");

    public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
            key("streaming-source.enable")
                    .booleanType()
                    .defaultValue(false)<commit_msg>[hotfix] fix typo in HiveOptions.<commit_after>/*
 * 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.connectors.hive;

import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.DescribedEnum;
import org.apache.flink.configuration.MemorySize;
import org.apache.flink.configuration.description.Description;
import org.apache.flink.configuration.description.InlineElement;
import org.apache.flink.connector.file.table.FileSystemConnectorOptions;

import java.time.Duration;

import static org.apache.flink.configuration.ConfigOptions.key;
import static org.apache.flink.configuration.description.TextElement.text;

/** This class holds configuration constants used by hive connector. */
public class HiveOptions {

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_READER =
            key("table.exec.hive.fallback-mapred-reader")
                    .booleanType()
                    .defaultValue(false)
                    .withDescription(
                            "If it is false, using flink native vectorized reader to read orc files; "
                                    + "If it is true, using hadoop mapred record reader to read orc files.");

    public static final ConfigOption<Boolean>
            TABLE_EXEC_HIVE_READ_PARTITION_WITH_SUBDIRECTORY_ENABLED =
                    key("table.exec.hive.read-partition-with-subdirectory.enabled")
                            .booleanType()
                            .defaultValue(true)
                            .withDescription(
                                    "If it is true, flink will read the files of partitioned hive table from subdirectories under the partition directory to be read.\n"
                                            + "If it is false, an exception that 'not a file: xxx' will be thrown when the partition directory contains any sub-directory.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM =
            key("table.exec.hive.infer-source-parallelism")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If is false, parallelism of source are set by config.\n"
                                    + "If is true, source parallelism is inferred according to splits number.\n");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_INFER_SOURCE_PARALLELISM_MAX =
            key("table.exec.hive.infer-source-parallelism.max")
                    .intType()
                    .defaultValue(1000)
                    .withDescription("Sets max infer parallelism for source operator.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_FALLBACK_MAPRED_WRITER =
            key("table.exec.hive.fallback-mapred-writer")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If it is false, using flink native writer to write parquet and orc files; "
                                    + "If it is true, using hadoop mapred record writer to write parquet and orc files.");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_LOAD_PARTITION_SPLITS_THREAD_NUM =
            key("table.exec.hive.load-partition-splits.thread-num")
                    .intType()
                    .defaultValue(3)
                    .withDescription(
                            "The thread number to split hive's partitions to splits. It should be bigger than 0.");

    public static final ConfigOption<MemorySize> TABLE_EXEC_HIVE_SPLIT_MAX_BYTES =
            key("table.exec.hive.split-max-size")
                    .memoryType()
                    .defaultValue(MemorySize.parse("128mb"))
                    .withDescription(
                            "The maximum number of bytes (default is 128MB) to pack into a split while reading Hive table. A split will be assigned to a reader. "
                                    + "It only works for the Hive table stored as ORC format.");

    public static final ConfigOption<MemorySize> TABLE_EXEC_HIVE_FILE_OPEN_COST =
            key("table.exec.hive.file-open-cost")
                    .memoryType()
                    .defaultValue(MemorySize.parse("4mb"))
                    .withDescription(
                            "The estimated cost (default is 4MB) to open a file. Used to split Hive's files to splits."
                                    + " When the value is over estimated, Flink will tend to pack Hive's data into less splits, which will be helpful when Hive's table contains many small files."
                                    + " And vice versa. It only works for the Hive table stored as ORC format.");

    public static final ConfigOption<Integer> TABLE_EXEC_HIVE_CALCULATE_PARTITION_SIZE_THREAD_NUM =
            key("table.exec.hive.calculate-partition-size.thread-num")
                    .intType()
                    .defaultValue(3)
                    .withDescription("The thread number to calculate partition's size.");

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_DYNAMIC_GROUPING_ENABLED =
            key("table.exec.hive.sink.sort-by-dynamic-partition.enable")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "Whether to enable sorting data by dynamic partition column or not when it's for dynamic partition writing."
                                    + " The default is to enable sorting."
                                    + " If it's enabled, the date will be sorted additionally by the dynamic partition columns"
                                    + " before writing into sink table."
                                    + " If it's disabled, there won't be extra sorting,"
                                    + " but it may throw OutOfMemory exception if there are too many dynamic partitions fall into same sink node."
                                    + " Note: it only works in batch mode.");

    /**
     * Hive users usually commit partition for metastore and a _SUCCESS file. That's why we create a
     * same option with {@link FileSystemConnectorOptions#SINK_PARTITION_COMMIT_POLICY_KIND} with
     * different default value.
     */
    public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_KIND =
            key("sink.partition-commit.policy.kind")
                    .stringType()
                    .defaultValue("metastore,success-file")
                    .withDescription(
                            "Policy to commit a partition is to notify the downstream"
                                    + " application that the partition has finished writing, the partition"
                                    + " is ready to be read."
                                    + " metastore: add partition to metastore. "
                                    + " success-file: add a success file to the partition directory. The success file name can be configured by the 'sink.partition-commit.success-file.name' option."
                                    + " Both can be configured at the same time: 'metastore,success-file'."
                                    + " custom: use policy class to create a commit policy."
                                    + " Support to configure multiple policies: 'metastore,success-file'.");

    public static final ConfigOption<String> SINK_PARTITION_COMMIT_POLICY_CLASS =
            FileSystemConnectorOptions.SINK_PARTITION_COMMIT_POLICY_CLASS;

    public static final ConfigOption<String> SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME =
            FileSystemConnectorOptions.SINK_PARTITION_COMMIT_SUCCESS_FILE_NAME;

    public static final ConfigOption<Boolean> TABLE_EXEC_HIVE_SINK_STATISTIC_AUTO_GATHER_ENABLE =
            key("table.exec.hive.sink.statistic-auto-gather.enable")
                    .booleanType()
                    .defaultValue(true)
                    .withDescription(
                            "If it's true, Flink will gather statistic automatically during writing Hive Table."
                                    + " If it's false, no any statistic will be gathered. The default value is true."
                                    + " For ORC and Parquet format, numFiles/totalSize/numRows/rawDataSize can be gathered."
                                    + " For other format, only numFiles/totalSize can be gathered."
                                    + " Note: only batch mode supports auto gather statistic, stream mode doesn't support it yet.");

    public static final ConfigOption<Integer>
            TABLE_EXEC_HIVE_SINK_STATISTIC_AUTO_GATHER_THREAD_NUM =
                    key("table.exec.hive.sink.statistic-auto-gather.thread-num")
                            .intType()
                            .defaultValue(3)
                            .withDescription(
                                    "The number of threads used to gather statistic during writing Hive Table"
                                            + " when the table is stored as ORC or Parquet format."
                                            + " The default value is 3.");

    public static final ConfigOption<Boolean> STREAMING_SOURCE_ENABLE =
            key("streaming-source.enable")
                    .booleanType()
                    .defaultValue(false)
{'fraction_non_alphanumeric': 0.066455078125, 'fraction_numerical': 0.00205078125, 'mean_word_length': 1.8140972794723824, 'pattern_counts': {'":': 0, '<': 35, '<?xml version=': 0, '>': 35, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 76, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1469851', 'n_tokens_mistral': 4941, 'n_tokens_neox': 4700, 'n_words': 1565}