more tables

delta-incubator · Dec 4, 2022 · 4c3c62b · 4c3c62b
1 parent de68fc8
commit 4c3c62b
Show file tree

Hide file tree

Showing 90 changed files with 263 additions and 116 deletions.
diff --git a/Makefile b/Makefile
@@ -25,7 +25,8 @@ lint-bandit: ## Run bandit
 	@echo "\n${BLUE}Running bandit...${NC}\n"
 	@${POETRY_RUN} bandit -r ${PROJ}
 
-lint-base: lint-flake8 lint-bandit ## Just run the linters without autolinting
+#lint-base: lint-flake8 lint-bandit ## Just run the linters without autolinting
+lint-base: lint-flake8 # TODO: Can we drop bandit?
 
 lint: autolint lint-base lint-mypy ## Autolint and code linting
 

diff --git a/dat/generated_tables.py b/dat/generated_tables.py
@@ -1,14 +1,14 @@
-from decimal import Decimal
 import os
+import random
 from datetime import date, datetime, timedelta
+from decimal import Decimal
 from pathlib import Path
-import random
 from typing import Callable, List, Tuple
 
-from delta.tables import DeltaTable
 import pyspark.sql
-from pyspark.sql import SparkSession
 import pyspark.sql.types as types
+from delta.tables import DeltaTable
+from pyspark.sql import SparkSession
 
 from dat.models import TableVersionMetadata, TestCaseInfo
 from dat.spark_builder import get_spark_session
@@ -158,16 +158,17 @@ def create_multi_partitioned(case: TestCaseInfo, spark: SparkSession):
 
 
 @reference_table(
-    name="multi_partitioned_2",
-    description="Multiple levels of partitioning, with boolean, timestamp, and decimal partition columns"
+    name='multi_partitioned_2',
+    description=('Multiple levels of partitioning, with boolean, timestamp, and '
+                 'decimal partition columns')
 )
 def create_multi_partitioned_2(case: TestCaseInfo, spark: SparkSession):
     columns = ['bool', 'time', 'amount', 'int']
     partition_columns = ['bool', 'time', 'amount']
     data = [
-        (True, datetime(1970, 1, 1), Decimal("200.00"), 1),
-        (True, datetime(1970, 1, 1, 12, 30), Decimal("200.00"), 2),
-        (False, datetime(1970, 1, 2, 8, 45),  Decimal("12.00"), 3)
+        (True, datetime(1970, 1, 1), Decimal('200.00'), 1),
+        (True, datetime(1970, 1, 1, 12, 30), Decimal('200.00'), 2),
+        (False, datetime(1970, 1, 2, 8, 45), Decimal('12.00'), 3)
     ]
     df = spark.createDataFrame(data, schema=columns)
     df.repartition(1).write.format('delta').partitionBy(
@@ -194,24 +195,25 @@ def with_schema_change(case: TestCaseInfo, spark: SparkSession):
         case.delta_root)
     save_expected(case)
 
+
 @reference_table(
     name='all_primitive_types',
     description='Table containing all non-nested types',
 )
 def create_all_primitive_types(case: TestCaseInfo, spark: SparkSession):
     schema = types.StructType([
-        types.StructField("utf8", types.StringType()),
-        types.StructField("int64", types.LongType()),
-        types.StructField("int32", types.IntegerType()),
-        types.StructField("int16", types.ShortType()),
-        types.StructField("int8", types.ByteType()),
-        types.StructField("float32", types.FloatType()),
-        types.StructField("float64", types.DoubleType()),
-        types.StructField("bool", types.BooleanType()),
-        types.StructField("binary", types.BinaryType()),
-        types.StructField("decimal", types.DecimalType(5, 3)),
-        types.StructField("date32", types.DateType()),
-        types.StructField("timestamp", types.TimestampType()),
+        types.StructField('utf8', types.StringType()),
+        types.StructField('int64', types.LongType()),
+        types.StructField('int32', types.IntegerType()),
+        types.StructField('int16', types.ShortType()),
+        types.StructField('int8', types.ByteType()),
+        types.StructField('float32', types.FloatType()),
+        types.StructField('float64', types.DoubleType()),
+        types.StructField('bool', types.BooleanType()),
+        types.StructField('binary', types.BinaryType()),
+        types.StructField('decimal', types.DecimalType(5, 3)),
+        types.StructField('date32', types.DateType()),
+        types.StructField('timestamp', types.TimestampType()),
     ])
 
     df = spark.createDataFrame([
@@ -225,7 +227,7 @@ def create_all_primitive_types(case: TestCaseInfo, spark: SparkSession):
             float(i),
             i % 2 == 0,
             bytes(i),
-            Decimal("10.000") + i,
+            Decimal('10.000') + i,
             date(1970, 1, 1) + timedelta(days=i),
             datetime(1970, 1, 1) + timedelta(hours=i)
         )
@@ -240,63 +242,140 @@ def create_all_primitive_types(case: TestCaseInfo, spark: SparkSession):
     description='Table containing various nested types',
 )
 def create_nested_types(case: TestCaseInfo, spark: SparkSession):
-    schema = types.StructType([
-        types.StructField("struct", types.StructType([
-            types.StructField("float64", types.DoubleType()),
-            types.StructField("bool", types.BooleanType()),
-        ])),
-        types.StructField("array", types.ArrayType(types.ShortType())),
-        types.StructField("map", types.MapType(types.StringType(), types.IntegerType())),
-    ])
+    schema = types.StructType([types.StructField(
+        'struct', types.StructType(
+            [types.StructField(
+                'float64', types.DoubleType()),
+             types.StructField(
+                'bool', types.BooleanType()), ])),
+        types.StructField(
+        'array', types.ArrayType(
+            types.ShortType())),
+        types.StructField(
+        'map', types.MapType(
+            types.StringType(),
+            types.IntegerType())), ])
 
     df = spark.createDataFrame([
         (
-            { "float64": float(i), "bool": i % 2 == 0 },
+            {'float64': float(i), 'bool': i % 2 == 0},
             list(range(i + 1)),
-            { str(i): i for i in range(i) }
+            {str(i): i for i in range(i)}
         )
         for i in range(5)
     ], schema=schema)
 
     df.repartition(1).write.format('delta').save(case.delta_root)
 
 
-def get_sample_data(spark: SparkSession, seed: int=42, nrows: int=5) -> pyspark.sql.DataFrame:
+def get_sample_data(
+        spark: SparkSession, seed: int = 42, nrows: int = 5) -> pyspark.sql.DataFrame:
     # Use seed to get consistent data between runs, for reproducibility
     random.seed(seed)
     return spark.createDataFrame([
         (
-            random.choice(["a", "b", "c", None]),
+            random.choice(['a', 'b', 'c', None]),
             random.randint(0, 1000),
             date(random.randint(1970, 2020), random.randint(1, 12), 1)
         )
         for i in range(nrows)
-    ], schema=["letter", "int", "date"])
+    ], schema=['letter', 'int', 'date'])
 
 
 @reference_table(
     name='with_checkpoint',
     description='Table with a checkpoint',
 )
 def create_with_checkpoint(case: TestCaseInfo, spark: SparkSession):
-    spark.conf.set("spark.databricks.delta.retentionDurationCheck.enabled", "false")
+    df = get_sample_data(spark)
+
+    (DeltaTable.create(spark)
+     .location(str(Path(case.delta_root).absolute()))
+     .addColumns(df.schema)
+     .property('delta.checkpointInterval', '2')
+     .execute())
+
+    for i in range(3):
+        df = get_sample_data(spark, seed=i, nrows=5)
+        df.repartition(1).write.format('delta').mode(
+            'overwrite').save(case.delta_root)
+
+    assert any(path.suffixes == ['.checkpoint', '.parquet']
+               for path in (Path(case.delta_root) / '_delta_log').iterdir())
+
+
+def remove_log_file(delta_root: str, version: int):
+    os.remove(os.path.join(delta_root, '_delta_log', f'{version:0>20}.json'))
+
+
+@reference_table(
+    name='no_replay',
+    description='Table with a checkpoint and prior commits cleaned up',
+)
+def create_no_replay(case: TestCaseInfo, spark: SparkSession):
+    spark.conf.set(
+        'spark.databricks.delta.retentionDurationCheck.enabled', 'false')
 
     df = get_sample_data(spark)
-    
+
     table = (DeltaTable.create(spark)
+             .location(str(Path(case.delta_root).absolute()))
+             .addColumns(df.schema)
+             .property('delta.checkpointInterval', '2')
+             .execute())
+
+    for i in range(3):
+        df = get_sample_data(spark, seed=i, nrows=5)
+        df.repartition(1).write.format('delta').mode(
+            'overwrite').save(case.delta_root)
+
+    table.vacuum(retentionHours=0)
+
+    remove_log_file(case.delta_root, version=0)
+    remove_log_file(case.delta_root, version=1)
+
+    files_in_log = list((Path(case.delta_root) / '_delta_log').iterdir())
+    assert any(path.suffixes == ['.checkpoint', '.parquet']
+               for path in files_in_log)
+    assert not any(path.name == f'{0:0>20}.json' for path in files_in_log)
+
+
+@reference_table(
+    name='stats_as_struct',
+    description='Table with stats only written as struct (not JSON) with Checkpoint',
+)
+def create_stats_as_struct(case: TestCaseInfo, spark: SparkSession):
+    df = get_sample_data(spark)
+    (DeltaTable.create(spark)
         .location(str(Path(case.delta_root).absolute()))
         .addColumns(df.schema)
-        .property("delta.checkpointInterval", "2")
-        .property("delta.logRetentionDuration", "0 days")
+        .property('delta.checkpointInterval', '2')
+        .property('delta.checkpoint.writeStatsAsStruct', 'true')
+        .property('delta.checkpoint.writeStatsAsJson', 'false')
         .execute())
 
-    for i in range(5):
+    for i in range(3):
         df = get_sample_data(spark, seed=i, nrows=5)
-        df.repartition(1).write.format('delta').mode('overwrite').save(case.delta_root)
-
-    assert any(path.suffixes == [".checkpoint", ".parquet"]
-               for path in  (Path(case.delta_root) / "_delta_log").iterdir())
+        df.repartition(1).write.format('delta').mode(
+            'overwrite').save(case.delta_root)
 
-    table.vacuum(retentionHours=0)
 
+@reference_table(
+    name='no_stats',
+    description='Table with no stats',
+)
+def create_no_stats(case: TestCaseInfo, spark: SparkSession):
+    df = get_sample_data(spark)
+    (DeltaTable.create(spark)
+        .location(str(Path(case.delta_root).absolute()))
+        .addColumns(df.schema)
+        .property('delta.checkpointInterval', '2')
+        .property('delta.checkpoint.writeStatsAsStruct', 'false')
+        .property('delta.checkpoint.writeStatsAsJson', 'false')
+        .property('delta.dataSkippingNumIndexedCols', '0')
+        .execute())
 
+    for i in range(3):
+        df = get_sample_data(spark, seed=i, nrows=5)
+        df.repartition(1).write.format('delta').mode(
+            'overwrite').save(case.delta_root)
diff --git a/dat/main.py b/dat/main.py
@@ -45,7 +45,8 @@ def write_generated_reference_tables(table_name: Optional[str]):
                 create_table()
                 break
         else:
-            raise ValueError(f"Could not find generated table named '{table_name}'")
+            raise ValueError(
+                f"Could not find generated table named '{table_name}'")
     else:
         out_base = Path('out/reader_tests/generated')
         shutil.rmtree(out_base)

diff --git a/out/reader_tests/generated/all_primitive_types/delta/_delta_log/00000000000000000000.json b/out/reader_tests/generated/all_primitive_types/delta/_delta_log/00000000000000000000.json
@@ -1,4 +1,4 @@
 {"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
-{"metaData":{"id":"90ceea80-98b2-4a16-99b4-bd8590371cc5","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"utf8\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int64\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int32\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int16\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int8\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float32\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float64\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"bool\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}},{\"name\":\"binary\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}},{\"name\":\"decimal\",\"type\":\"decimal(5,3)\",\"nullable\":true,\"metadata\":{}},{\"name\":\"date32\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}},{\"name\":\"timestamp\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1668830585678}}
-{"add":{"path":"part-00000-9a4d4f91-4cab-4294-9e26-97785c9556fc-c000.snappy.parquet","partitionValues":{},"size":3206,"modificationTime":1668830585964,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"utf8\":\"0\",\"int64\":0,\"int32\":0,\"int16\":0,\"int8\":0,\"float32\":0.0,\"float64\":0.0,\"decimal\":10.000,\"date32\":\"1970-01-01\",\"timestamp\":\"1970-01-01T00:00:00.000-08:00\"},\"maxValues\":{\"utf8\":\"4\",\"int64\":4,\"int32\":4,\"int16\":4,\"int8\":4,\"float32\":4.0,\"float64\":4.0,\"decimal\":14.000,\"date32\":\"1970-01-05\",\"timestamp\":\"1970-01-01T04:00:00.000-08:00\"},\"nullCount\":{\"utf8\":0,\"int64\":0,\"int32\":0,\"int16\":0,\"int8\":0,\"float32\":0,\"float64\":0,\"bool\":0,\"binary\":0,\"decimal\":0,\"date32\":0,\"timestamp\":0}}"}}
-{"commitInfo":{"timestamp":1668830585986,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"5","numOutputBytes":"3206"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"46722e06-4b7f-463f-8608-94b227d9c98c"}}
+{"metaData":{"id":"0f4744d1-b3e6-48b4-b484-292858112c44","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"utf8\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int64\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int32\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int16\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"int8\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float32\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"float64\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"bool\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}},{\"name\":\"binary\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}},{\"name\":\"decimal\",\"type\":\"decimal(5,3)\",\"nullable\":true,\"metadata\":{}},{\"name\":\"date32\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}},{\"name\":\"timestamp\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1668830722564}}
+{"add":{"path":"part-00000-2aa4115c-ce78-44bc-9cb8-9b509ffe48cd-c000.snappy.parquet","partitionValues":{},"size":3206,"modificationTime":1668830722824,"dataChange":true,"stats":"{\"numRecords\":5,\"minValues\":{\"utf8\":\"0\",\"int64\":0,\"int32\":0,\"int16\":0,\"int8\":0,\"float32\":0.0,\"float64\":0.0,\"decimal\":10.000,\"date32\":\"1970-01-01\",\"timestamp\":\"1970-01-01T00:00:00.000-08:00\"},\"maxValues\":{\"utf8\":\"4\",\"int64\":4,\"int32\":4,\"int16\":4,\"int8\":4,\"float32\":4.0,\"float64\":4.0,\"decimal\":14.000,\"date32\":\"1970-01-05\",\"timestamp\":\"1970-01-01T04:00:00.000-08:00\"},\"nullCount\":{\"utf8\":0,\"int64\":0,\"int32\":0,\"int16\":0,\"int8\":0,\"float32\":0,\"float64\":0,\"bool\":0,\"binary\":0,\"decimal\":0,\"date32\":0,\"timestamp\":0}}"}}
+{"commitInfo":{"timestamp":1668830722845,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"5","numOutputBytes":"3206"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"d3b15c24-3d91-4597-9d83-6f514b45b92c"}}
diff --git a/...294-9e26-97785c9556fc-c000.snappy.parquet → ...4bc-9cb8-9b509ffe48cd-c000.snappy.parquet b/...294-9e26-97785c9556fc-c000.snappy.parquet → ...4bc-9cb8-9b509ffe48cd-c000.snappy.parquet
diff --git a/out/reader_tests/generated/basic_append/delta/_delta_log/00000000000000000000.json b/out/reader_tests/generated/basic_append/delta/_delta_log/00000000000000000000.json
@@ -1,4 +1,4 @@
 {"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
-{"metaData":{"id":"d379f8f1-60b4-456c-a941-20ac267cbea3","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"letter\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"number\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"a_float\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1668830560516}}
-{"add":{"path":"part-00000-7ced08f4-9725-4547-a3dd-1e39a1eaf7b0-c000.snappy.parquet","partitionValues":{},"size":996,"modificationTime":1668830562714,"dataChange":true,"stats":"{\"numRecords\":3,\"minValues\":{\"letter\":\"a\",\"number\":1,\"a_float\":1.1},\"maxValues\":{\"letter\":\"c\",\"number\":3,\"a_float\":3.3},\"nullCount\":{\"letter\":0,\"number\":0,\"a_float\":0}}"}}
-{"commitInfo":{"timestamp":1668830562838,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"3","numOutputBytes":"996"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"92a218de-ba83-4830-b62a-caa0ee1e35ad"}}
+{"metaData":{"id":"40c6ddf2-93cd-414e-92fa-3496d3368465","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"letter\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"number\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"a_float\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1668830697027}}
+{"add":{"path":"part-00000-e3224358-9a1b-47d1-a581-6f0b434b2fc6-c000.snappy.parquet","partitionValues":{},"size":996,"modificationTime":1668830699344,"dataChange":true,"stats":"{\"numRecords\":3,\"minValues\":{\"letter\":\"a\",\"number\":1,\"a_float\":1.1},\"maxValues\":{\"letter\":\"c\",\"number\":3,\"a_float\":3.3},\"nullCount\":{\"letter\":0,\"number\":0,\"a_float\":0}}"}}
+{"commitInfo":{"timestamp":1668830699482,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"3","numOutputBytes":"996"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"924fed05-58ad-4fab-92e4-e3b059d63c73"}}
diff --git a/out/reader_tests/generated/basic_append/delta/_delta_log/00000000000000000001.json b/out/reader_tests/generated/basic_append/delta/_delta_log/00000000000000000001.json
@@ -1,2 +1,2 @@
-{"add":{"path":"part-00000-127f8ce7-badc-4b8e-b585-674644c96f90-c000.snappy.parquet","partitionValues":{},"size":984,"modificationTime":1668830568044,"dataChange":true,"stats":"{\"numRecords\":2,\"minValues\":{\"letter\":\"d\",\"number\":4,\"a_float\":4.4},\"maxValues\":{\"letter\":\"e\",\"number\":5,\"a_float\":5.5},\"nullCount\":{\"letter\":0,\"number\":0,\"a_float\":0}}"}}
-{"commitInfo":{"timestamp":1668830568065,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"2","numOutputBytes":"984"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"53ffff03-f331-461a-aa26-0e1aa244f2ca"}}
+{"add":{"path":"part-00000-5f3baff2-c3a8-4b38-ba53-cd71291b3fbd-c000.snappy.parquet","partitionValues":{},"size":984,"modificationTime":1668830704664,"dataChange":true,"stats":"{\"numRecords\":2,\"minValues\":{\"letter\":\"d\",\"number\":4,\"a_float\":4.4},\"maxValues\":{\"letter\":\"e\",\"number\":5,\"a_float\":5.5},\"nullCount\":{\"letter\":0,\"number\":0,\"a_float\":0}}"}}
+{"commitInfo":{"timestamp":1668830704686,"operation":"WRITE","operationParameters":{"mode":"Append","partitionBy":"[]"},"readVersion":0,"isolationLevel":"Serializable","isBlindAppend":true,"operationMetrics":{"numFiles":"1","numOutputRows":"2","numOutputBytes":"984"},"engineInfo":"Apache-Spark/3.3.1 Delta-Lake/2.1.1","txnId":"abca60bd-c6e3-4b1f-bce7-f4ddef5f7f2f"}}
diff --git a/...b8e-b585-674644c96f90-c000.snappy.parquet → ...b38-ba53-cd71291b3fbd-c000.snappy.parquet b/...b8e-b585-674644c96f90-c000.snappy.parquet → ...b38-ba53-cd71291b3fbd-c000.snappy.parquet
diff --git a/...547-a3dd-1e39a1eaf7b0-c000.snappy.parquet → ...7d1-a581-6f0b434b2fc6-c000.snappy.parquet b/...547-a3dd-1e39a1eaf7b0-c000.snappy.parquet → ...7d1-a581-6f0b434b2fc6-c000.snappy.parquet