Insert only impala load template

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/impala_plugin.py

@@ -98,6 +98,10 @@ def initialize_job(self, context: JobContext) -> None:
             "periodic_snapshot", pathlib.Path(get_job_path("load/fact/snapshot"))
         )
 
+        context.templates.add_template(
+            "insert", pathlib.Path(get_job_path("load/fact/insert"))
+        )
+
         context.templates.add_template(
             "load/versioned", pathlib.Path(get_job_path("load/versioned"))
         )

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/00-fact-snapshot-definition.py

@@ -0,0 +1,25 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from pydantic import BaseModel
+from vdk.api.job_input import IJobInput
+from vdk.plugin.impala.templates.template_arguments_validator import (
+    TemplateArgumentsValidator,
+)
+
+
+class FactDailySnapshotParams(BaseModel):
+    target_schema: str
+    target_table: str
+    source_schema: str
+    source_view: str
+
+
+class FactDailySnapshot(TemplateArgumentsValidator):
+    TemplateParams = FactDailySnapshotParams
+
+    def __init__(self) -> None:
+        super().__init__()
+
+
+def run(job_input: IJobInput):
+    FactDailySnapshot().get_validated_args(job_input, job_input.get_arguments())

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/01-test-if-view-matches-target.sql

@@ -0,0 +1,3 @@
+SELECT * FROM {source_schema}.{source_view} LIMIT 0
+UNION ALL
+SELECT * FROM {target_schema}.{target_table} LIMIT 0;

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/02-insert-into-target.sql

@@ -0,0 +1,7 @@
+-- /* +SHUFFLE */ below is a query hint to Impala.
+-- Do not remove! https://www.cloudera.com/documentation/enterprise/5-9-x/topics/impala_hints.html
+INSERT INTO TABLE {target_schema}.{target_table} {_vdk_template_insert_partition_clause} /* +SHUFFLE */
+(
+  SELECT *
+  FROM   {source_schema}.{source_view}
+);

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/03-refresh.sql

@@ -0,0 +1 @@
+REFRESH {target_schema}.{target_table};

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/04-compute-stats.sql

@@ -0,0 +1 @@
+COMPUTE STATS {target_schema}.{target_table};

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/README.md

@@ -0,0 +1,43 @@
+### Purpose:
+
+This template can be used to load raw data from Data Lake to target Table in Data Warehouse. In summary, it appends all records from the source table to the target table.
+
+### Template Name (template_name):
+
+- "insert"
+
+### Template Parameters (template_args):
+
+- target_schema   - Data Warehouse schema, where target data is loaded
+- target_table    - Data Warehouse table of DW type table, where target data is loaded
+- source_schema   - Data Lake schema, where source raw data is loaded from
+- source_view     - Data Lake view, where source raw data is loaded from
+
+### Prerequisites:
+
+In order to use this template you need to ensure the following:
+- {source_schema}.{source_view} exists
+- {target_schema}.{target_table} exists
+- {source_schema}.{source_view} has the exact same schema as {target_schema}.{target_table}
+
+### Sample Usage:
+
+Say there is SDDC-related 'Snapshot Periodic Fact Table' called 'fact_vmc_utilization_cpu_mem_every5min_daily' in 'history' schema.
+Updating it with the latest raw data from a Data Lake (from source view called 'vw_fact_vmc_utilization_cpu_mem_every5min_daily' in 'default' schema) is done in the following manner:
+
+```python
+def run(job_input):
+    # . . .
+    template_args = {
+        'source_schema': 'default',
+        'source_view': 'vw_fact_vmc_utilization_cpu_mem_every5min_daily',
+        'target_schema': 'history',
+        'target_table': 'fact_vmc_utilization_cpu_mem_every5min_daily'
+    }
+    job_input.execute_template('insert', template_args)
+    # . . .
+```
+
+### Example
+
+See all templates in [our example documentation](https://github.com/vmware/versatile-data-kit/wiki/SQL-Data-Processing-templates-examples).

projects/vdk-plugins/vdk-impala/src/vdk/plugin/impala/templates/load/fact/insert/params.json

@@ -0,0 +1,6 @@
+[
+  "target_schema",
+  "target_table",
+  "source_schema",
+  "source_view"
+]

projects/vdk-plugins/vdk-impala/tests/functional/jobs/insert_template_job/01_prepare_input_data.py

@@ -0,0 +1,88 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.job_input import IJobInput
+
+
+__author__ = "VMware, Inc."
+__copyright__ = (
+    "Copyright 2019 VMware, Inc.  All rights reserved. -- VMware Confidential"
+)
+
+
+def run(job_input: IJobInput) -> None:
+    # Step 1: create a table that represents the current state
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{target_schema}`.`{target_table}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{target_schema}`.`{target_table}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT,
+          `cluster_count` BIGINT
+        ) STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{target_schema}`.`{target_table}` VALUES (
+          ("sddc01-r01", 1, "2019-11-18", 5 , 1),
+          ("sddc02-r01", 2, "2019-11-18", 4 , 1)
+        )
+    """
+    )
+
+    # Step 2: create a table that represents the next snapshot
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{source_schema}`.`{source_view}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{source_schema}`.`{source_view}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT,
+          `cluster_count` BIGINT
+        ) STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{source_schema}`.`{source_view}` VALUES (
+          ("sddc01-r01", 1, "2019-11-19", 5 , 1),
+          ("sddc01-r01", 1, "2019-11-20", 10, 2)
+        )
+    """
+    )
+
+    # Step 3: Create a table containing the state expected after updating the current state with the next snapshot
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{expect_schema}`.`{expect_table}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{expect_schema}`.`{expect_table}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT,
+          `cluster_count` BIGINT
+        ) STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{expect_schema}`.`{expect_table}` VALUES (
+          ("sddc01-r01", 1, "2019-11-18", 5 , 1),
+          ("sddc02-r01", 2, "2019-11-18", 4 , 1),
+          ("sddc01-r01", 1, "2019-11-19", 5 , 1),
+          ("sddc01-r01", 1, "2019-11-20", 10, 2)
+        )
+    """
+    )

projects/vdk-plugins/vdk-impala/tests/functional/jobs/insert_template_job/02_run_insert_template.py

@@ -0,0 +1,15 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.job_input import IJobInput
+
+__author__ = "VMware, Inc."
+__copyright__ = (
+    "Copyright 2019 VMware, Inc.  All rights reserved. -- VMware Confidential"
+)
+
+
+def run(job_input: IJobInput) -> None:
+    job_input.execute_template(
+        template_name="insert",
+        template_args=job_input.get_arguments(),
+    )

projects/vdk-plugins/vdk-impala/tests/functional/jobs/insert_template_partition_job/01_prepare_input_data.py

@@ -0,0 +1,92 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.job_input import IJobInput
+
+
+__author__ = "VMware, Inc."
+__copyright__ = (
+    "Copyright 2019 VMware, Inc.  All rights reserved. -- VMware Confidential"
+)
+
+
+def run(job_input: IJobInput) -> None:
+    # Step 1: create a table that represents the current стате
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{target_schema}`.`{target_table}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{target_schema}`.`{target_table}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT
+        ) PARTITIONED BY (`cluster_count` BIGINT) STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        TRUNCATE `{target_schema}`.`{target_table}`
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{target_schema}`.`{target_table}` PARTITION (cluster_count)  VALUES (
+          ("sddc01-r01", 1, "2019-11-18", 5 , 1),
+          ("sddc02-r01", 2, "2019-11-18", 4 , 1)
+        )
+    """
+    )
+
+    # Step 2: create a table that represents the next snapshot
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{source_schema}`.`{source_view}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{source_schema}`.`{source_view}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT,
+          `cluster_count` BIGINT
+        )  STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{source_schema}`.`{source_view}` VALUES (
+          ("sddc01-r01", 1, "2019-11-19", 5 , 1),
+          ("sddc01-r01", 1, "2019-11-20", 10 , 2)
+        )
+    """
+    )
+
+    # Step 3: Create a table containing the state expected after updating the current state with the next snapshot
+
+    # job_input.execute_query(u'''
+    #     DROP TABLE IF EXISTS `{expect_schema}`.`{expect_table}`
+    # ''')
+    job_input.execute_query(
+        """
+        CREATE TABLE IF NOT EXISTS `{expect_schema}`.`{expect_table}` (
+          `dim_sddc_sk` STRING,
+          `dim_org_id` INT,
+          `dim_date_id` TIMESTAMP,
+          `host_count` BIGINT,
+          `cluster_count` BIGINT
+        ) STORED AS PARQUET
+    """
+    )
+    job_input.execute_query(
+        """
+        INSERT OVERWRITE TABLE `{expect_schema}`.`{expect_table}` VALUES (
+          ("sddc01-r01", 1, "2019-11-18", 5 , 1),
+          ("sddc02-r01", 2, "2019-11-18", 4 , 1),
+          ("sddc01-r01", 1, "2019-11-19", 5 , 1),
+          ("sddc01-r01", 1, "2019-11-20", 10 , 2)
+        )
+    """
+    )

projects/vdk-plugins/vdk-impala/tests/functional/jobs/insert_template_partition_job/02_run_insert_template.py

@@ -0,0 +1,15 @@
+# Copyright 2021 VMware, Inc.
+# SPDX-License-Identifier: Apache-2.0
+from vdk.api.job_input import IJobInput
+
+__author__ = "VMware, Inc."
+__copyright__ = (
+    "Copyright 2019 VMware, Inc.  All rights reserved. -- VMware Confidential"
+)
+
+
+def run(job_input: IJobInput) -> None:
+    job_input.execute_template(
+        template_name="insert",
+        template_args=job_input.get_arguments(),
+    )

projects/vdk-plugins/vdk-impala/tests/functional/template_regression_test.py

@@ -596,3 +596,63 @@ def just_throw(*_, **kwargs):
                     "#parquet_ddl"
                 ),
             )
+
+    def test_insert(self) -> None:
+        test_schema = "vdkprototypes"
+        source_view = "vw_fact_vmc_utilization_cpu_mem_every5min_daily"
+        target_table = "dw_fact_vmc_utilization_cpu_mem_every5min_daily"
+        expect_table = "ex_fact_vmc_utilization_cpu_mem_every5min_daily"
+
+        res = self._run_job(
+            "insert_template_job",
+            {
+                "source_schema": test_schema,
+                "source_view": source_view,
+                "target_schema": test_schema,
+                "target_table": target_table,
+                "expect_schema": test_schema,
+                "expect_table": expect_table,
+            },
+        )
+        assert not res.exception
+
+        actual_rs = self._run_query(f"SELECT * FROM {test_schema}.{target_table}")
+        expected_rs = self._run_query(f"SELECT * FROM {test_schema}.{expect_table}")
+        assert actual_rs.output and expected_rs.output
+
+        actual = {x for x in actual_rs.output.split("\n")}
+        expected = {x for x in expected_rs.output.split("\n")}
+
+        self.assertSetEqual(
+            actual, expected, f"Elements in {expect_table} and {target_table} differ."
+        )
+
+    def test_insert_partition(self) -> None:
+        test_schema = "vdkprototypes"
+        source_view = "vw_fact_vmc_utilization_cpu_mem_every5min_daily_partition"
+        target_table = "dw_fact_vmc_utilization_cpu_mem_every5min_daily_partition"
+        expect_table = "ex_fact_vmc_utilization_cpu_mem_every5min_daily_partition"
+
+        res = self._run_job(
+            "insert_template_partition_job",
+            {
+                "source_schema": test_schema,
+                "source_view": source_view,
+                "target_schema": test_schema,
+                "target_table": target_table,
+                "expect_schema": test_schema,
+                "expect_table": expect_table,
+            },
+        )
+        assert not res.exception
+
+        actual_rs = self._run_query(f"SELECT * FROM {test_schema}.{target_table}")
+        expected_rs = self._run_query(f"SELECT * FROM {test_schema}.{expect_table}")
+        assert actual_rs.output and expected_rs.output
+
+        actual = {x for x in actual_rs.output.split("\n")}
+        expected = {x for x in expected_rs.output.split("\n")}
+
+        self.assertSetEqual(
+            actual, expected, f"Elements in {expect_table} and {target_table} differ."
+        )