Package io.delta.kernel.internal.util

Class PartitionUtils

Object
io.delta.kernel.internal.util.PartitionUtils
public class PartitionUtils extends Object

  • Method Details

    • withPartitionColumns

      public static ColumnarBatch withPartitionColumns(ColumnarBatch dataBatch, StructType logicalReadSchema, Map<String,String> partitionValues, ExpressionHandler expressionHandler)
      Utility method to attach partition columns to the given data batch.
      Parameters:
      dataBatch - Data batch to which the partition columns will be added.
      logicalReadSchema - Logical schema of the table scan. Used to insert partition columns at the right positions in the data batch. This logical schema must contain column mapping metadata if column mapping is enabled.
      partitionValues - Map of partition column name to value.
      expressionHandler - Expression handler used to evaluate the partition values.
      Returns:
      A new ColumnarBatch with the partition columns added.

    • serializePartitionMap

      public static MapValue serializePartitionMap(Map<String,Literal> partitionValueMap)
      Convert the given partition values to a MapValue that can be serialized to a Delta commit file.
      Parameters:
      partitionValueMap - Expected the partition column names to be same case as in the schema. We want to preserve the case of the partition column names when serializing to the Delta commit file.
      Returns:
      MapValue representing the serialized partition values that can be written to a Delta commit file.

    • validateAndSanitizePartitionValues

      public static Map<String,Literal> validateAndSanitizePartitionValues(StructType tableSchema, List<String> partitionColNames, Map<String,Literal> partitionValues)
      Validate partitionValues contains values for every partition column in the table and the type of the value is correct. Once validated the partition values are sanitized to match the case of the partition column names in the table schema and returned
      Parameters:
      tableSchema - Schema of the table.
      partitionColNames - Partition column name. These should be from the table metadata that retain the same case as in the table schema.
      partitionValues - Map of partition column to value map given by the connector
      Returns:
      Sanitized partition values.

    • validatePredicateOnlyOnPartitionColumns

      public static void validatePredicateOnlyOnPartitionColumns(Predicate predicate, Set<String> partitionColNames)
      Validate that the given predicate references only (and at least one) partition columns.
      Throws:
      IllegalArgumentException - if the predicate does not reference any partition columns or if it references any data columns

    • splitMetadataAndDataPredicates

      public static Tuple2<Predicate,Predicate> splitMetadataAndDataPredicates(Predicate predicate, Set<String> partitionColNames)
      Split the given predicate into predicate on partition columns and predicate on data columns.
      Parameters:
      predicate -
      partitionColNames -
      Returns:
      Tuple of partition column predicate and data column predicate.

    • rewritePartitionPredicateOnCheckpointFileSchema

      public static Predicate rewritePartitionPredicateOnCheckpointFileSchema(Predicate predicate, Map<String,StructField> partitionColNameToField)
      Rewrite the given predicate on partition columns on `partitionValues_parsed` in checkpoint schema. The rewritten predicate can be pushed to the Parquet reader when reading the checkpoint files.
      Parameters:
      predicate - Predicate on partition columns.
      partitionColNameToField - Map of partition column name (in lower case) to its StructField.
      Returns:
      Rewritten Predicate on `partitionValues_parsed` in `add`.

    • rewritePartitionPredicateOnScanFileSchema

      public static Predicate rewritePartitionPredicateOnScanFileSchema(Predicate predicate, Map<String,StructField> partitionColMetadata)
      Utility method to rewrite the partition predicate referring to the table schema as predicate referring to the partitionValues in scan files read from Delta log. The scan file batch is returned by the Scan.getScanFiles(Engine).

      E.g. given predicate on partition columns: p1 = 'new york' && p2 >= 26 where p1 is of type string and p2 is of int Rewritten expression looks like: element_at(Column('add', 'partitionValues'), 'p1') = 'new york' && partition_value(element_at(Column('add', 'partitionValues'), 'p2'), 'integer') >= 26

      The column `add.partitionValues` is a map(string -> string) type. Each partition values is in string serialization format according to the Delta protocol. Expression `partition_value` deserializes the string value into the given partition column type value. String type partition values don't need any deserialization.

      Parameters:
      predicate - Predicate containing filters only on partition columns.
      partitionColMetadata - Map of partition column name (in lower case) to its type.
      Returns:

    • getTargetDirectory

      public static String getTargetDirectory(String dataRoot, List<String> partitionColNames, Map<String,Literal> partitionValues)
      Get the target directory for writing data for given partition values. Example: Given partition values (part1=1, part2='abc'), the target directory will be for a table rooted at 's3://bucket/table': 's3://bucket/table/part1=1/part2=abc'.
      Parameters:
      dataRoot - Root directory where the data is stored.
      partitionColNames - Partition column names. We need this to create the target directory structure that is consistent levels of directories.
      partitionValues - Partition values to create the target directory.
      Returns:
      Target directory path.

    • tryParseTimestamp

      public static long tryParseTimestamp(String partitionValue)
      Try parsing the timestamp, could be in the standard format or ISO8601 format. Return the Literal Object.