changing config name and defaulting to lookback of 1

metadata-ingestion/src/datahub/ingestion/source/delta_lake/config.py

@@ -59,9 +59,9 @@ class Config:
         default=AllowDenyPattern.allow_all(),
         description="regex patterns for tables to filter in ingestion.",
     )
-    version_history_limit: Optional[int] = Field(
-        default=None,
-        description="Number of previous version histories to be ingested. If set to None all version history will be ingested.",
+    version_history_lookback: Optional[int] = Field(
+        default=1,
+        description="Number of previous version histories to be ingested. Defaults to 1. If set to -1 all version history will be ingested.",
     )
 
     s3: Optional[S3] = Field()
@@ -76,6 +76,12 @@ def is_s3(self):
     def get_complete_path(self):
         return self._complete_path
 
+    @pydantic.validator("version_history_lookback")
+    def negative_version_history_implies_no_limit(cls, v):
+        if v and v < 0:
+            return None
+        return v
+
     @pydantic.root_validator()
     def validate_config(cls, values: Dict) -> Dict[str, Any]:
         values["_is_s3"] = is_s3_uri(values.get("base_path",""))

metadata-ingestion/src/datahub/ingestion/source/delta_lake/source.py

@@ -144,7 +144,7 @@ def get_fields(self, delta_table: DeltaTable) -> List[SchemaField]:
     def _create_operation_aspect_wu(
         self, delta_table: DeltaTable, dataset_urn: str
     ) -> Iterable[MetadataWorkUnit]:
-        for hist in delta_table.history(limit=self.source_config.version_history_limit):
+        for hist in delta_table.history(limit=self.source_config.version_history_lookback):
 
             # History schema picked up from https://docs.delta.io/latest/delta-utility.html#retrieve-delta-table-history
             reported_time: int = int(time.time() * 1000)

metadata-ingestion/tests/integration/delta_lake/sources/local/allow_table.json

@@ -6,6 +6,7 @@
     "platform_instance": "my-platform",
     "table_pattern": {
       "allow": ["s*"]
-    }
+    },
+    "version_history_lookback": -1
   }
-}
+}

metadata-ingestion/tests/integration/delta_lake/sources/local/inner_table.json

@@ -2,6 +2,7 @@
   "type": "delta-lake",
   "config": {
     "env": "UAT",
-    "base_path": "tests/integration/delta_lake/test_data/delta_tables"
+    "base_path": "tests/integration/delta_lake/test_data/delta_tables",
+    "version_history_lookback": -1
   }
-}
+}

metadata-ingestion/tests/integration/delta_lake/sources/local/relative_path.json

@@ -3,6 +3,7 @@
   "config": {
     "env": "UAT",
     "base_path": "tests/integration/delta_lake/test_data/",
-    "relative_path":"delta_tables/my_table_basic/"
+    "relative_path":"delta_tables/my_table_basic/",
+    "version_history_lookback": -1
   }
 }

metadata-ingestion/tests/integration/delta_lake/sources/local/single_table.json

@@ -2,7 +2,6 @@
   "type": "delta-lake",
   "config": {
     "env": "UAT",
-    "base_path": "tests/integration/delta_lake/test_data/delta_tables/my_table_basic",
-    "version_history_limit": 1
+    "base_path": "tests/integration/delta_lake/test_data/delta_tables/my_table_basic"
   }
-}
+}