From 0fa8ddb7e8b083a95cea43fef2f109a6400af7b2 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Tue, 19 Sep 2023 15:44:51 +0200
Subject: [PATCH 01/10] SNFLK autocast types when using string stage table

---
 .../php-datatypes/src/Definition/Common.php   |  11 +-
 .../src/Definition/DefinitionInterface.php    |  10 ++
 .../Snowflake/SnowflakeImportOptions.php      |  18 ++-
 .../Snowflake/ToFinalTable/SqlBuilder.php     | 104 +++++++++++----
 .../Backend/SourceDestinationColumnMap.php    |  91 +++++++++++++
 .../Snowflake/ToFinal/FullImportTest.php      |  93 ++++++++++++++
 .../ToFinal/IncrementalImportTest.php         | 121 ++++++++++++++++++
 7 files changed, 420 insertions(+), 28 deletions(-)
 create mode 100644 packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php

diff --git a/packages/php-datatypes/src/Definition/Common.php b/packages/php-datatypes/src/Definition/Common.php
index 0d421b813..4bab0f25d 100644
--- a/packages/php-datatypes/src/Definition/Common.php
+++ b/packages/php-datatypes/src/Definition/Common.php
@@ -93,10 +93,12 @@ public function toMetadata(): array
             [
                 'key' => self::KBC_METADATA_KEY_TYPE,
                 'value' => $this->getType(),
-            ],[
+            ],
+            [
                 'key' => self::KBC_METADATA_KEY_NULLABLE,
                 'value' => $this->isNullable(),
-            ],[
+            ],
+            [
                 'key' => self::KBC_METADATA_KEY_BASETYPE,
                 'value' => $this->getBasetype(),
             ],
@@ -166,4 +168,9 @@ protected function validateMaxLength($length, int $max, int $min = 1): bool
         }
         return (int) $length >= $min && (int) $length <= $max;
     }
+
+    public function isSameType(DefinitionInterface $definition): bool
+    {
+        return $this->type === $definition->getType();
+    }
 }
diff --git a/packages/php-datatypes/src/Definition/DefinitionInterface.php b/packages/php-datatypes/src/Definition/DefinitionInterface.php
index d328cb3dc..385bbad8e 100644
--- a/packages/php-datatypes/src/Definition/DefinitionInterface.php
+++ b/packages/php-datatypes/src/Definition/DefinitionInterface.php
@@ -15,5 +15,15 @@ public function toArray(): array;
 
     public function getBasetype(): string;
 
+    public function getType(): string;
+
+    public function getLength(): ?string;
+
+    public function isNullable(): bool;
+
+    public function getDefault(): ?string;
+
     public static function getTypeByBasetype(string $basetype): string;
+
+    public function isSameType(DefinitionInterface $definition): bool;
 }
diff --git a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
index 9446a8908..896e432e6 100644
--- a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
+++ b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
@@ -8,6 +8,11 @@
 
 class SnowflakeImportOptions extends ImportOptions
 {
+    /**
+     * @var string[]
+     */
+    private array $autoCastTypes;
+
     /** @var self::SAME_TABLES_* */
     private bool $requireSameTables;
 
@@ -19,6 +24,7 @@ class SnowflakeImportOptions extends ImportOptions
      * @param self::SAME_TABLES_* $requireSameTables
      * @param self::NULL_MANIPULATION_* $nullManipulation
      * @param string[] $ignoreColumns
+     * @param string[] $autoCastTypes
      */
     public function __construct(
         array $convertEmptyValuesToNull = [],
@@ -27,7 +33,8 @@ public function __construct(
         int $numberOfIgnoredLines = 0,
         bool $requireSameTables = self::SAME_TABLES_NOT_REQUIRED,
         bool $nullManipulation = self::NULL_MANIPULATION_ENABLED,
-        array $ignoreColumns = []
+        array $ignoreColumns = [],
+        array $autoCastTypes = [],
     ) {
         parent::__construct(
             $convertEmptyValuesToNull,
@@ -39,6 +46,7 @@ public function __construct(
         );
         $this->requireSameTables = $requireSameTables;
         $this->nullManipulation = $nullManipulation;
+        $this->autoCastTypes = $autoCastTypes;
     }
 
     public function isRequireSameTables(): bool
@@ -50,4 +58,12 @@ public function isNullManipulationEnabled(): bool
     {
         return $this->nullManipulation === self::NULL_MANIPULATION_ENABLED;
     }
+
+    /**
+     * @return string[]
+     */
+    public function autoCastTypes(): array
+    {
+        return $this->autoCastTypes;
+    }
 }
diff --git a/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php b/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
index a832f88db..24f45108b 100644
--- a/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
+++ b/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
@@ -8,12 +8,9 @@
 use Keboola\Datatype\Definition\Snowflake;
 use Keboola\Db\ImportExport\Backend\Snowflake\Helper\QuoteHelper;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
+use Keboola\Db\ImportExport\Backend\SourceDestinationColumnMap;
 use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
-use Keboola\Db\ImportExport\ImportOptionsInterface;
-use Keboola\Db\ImportExport\Storage\Snowflake\Table;
-use Keboola\Db\ImportExport\Storage\SourceInterface;
 use Keboola\TableBackendUtils\Column\Snowflake\SnowflakeColumn;
-use Keboola\TableBackendUtils\Escaping\Exasol\ExasolQuote;
 use Keboola\TableBackendUtils\Escaping\Snowflake\SnowflakeQuote;
 use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableDefinition;
 
@@ -165,6 +162,11 @@ public function getInsertAllIntoTargetTableCommand(
         SnowflakeImportOptions $importOptions,
         string $timestamp
     ): string {
+        $columnMap = SourceDestinationColumnMap::createForTables(
+            $sourceTableDefinition,
+            $destinationTableDefinition,
+            $importOptions->ignoreColumns()
+        );
         $destinationTable = sprintf(
             '%s.%s',
             SnowflakeQuote::quoteSingleIdentifier($destinationTableDefinition->getSchemaName()),
@@ -184,44 +186,67 @@ public function getInsertAllIntoTargetTableCommand(
 
         $columnsSetSql = [];
 
-        /** @var SnowflakeColumn $columnDefinition */
-        foreach ($sourceTableDefinition->getColumnsDefinitions() as $columnDefinition) {
+        /** @var SnowflakeColumn $sourceColumn */
+        foreach ($sourceTableDefinition->getColumnsDefinitions() as $sourceColumn) {
             // output mapping same tables are required do not convert nulls to empty strings
             if (!$importOptions->isNullManipulationEnabled()) {
-                $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName());
+                $destinationColumn = $columnMap->getDestination($sourceColumn);
+                $type = $destinationColumn->getColumnDefinition()->getType();
+                $useAutoCast = in_array($type, $importOptions->autoCastTypes(), true);
+                $isSameType = $type === $sourceColumn->getColumnDefinition()->getType();
+                if ($useAutoCast && !$isSameType) {
+                    if ($type === Snowflake::TYPE_OBJECT) {
+                        // object can't be casted from string but can be casted from variant
+                        $columnsSetSql[] = sprintf(
+                            'CAST(TO_VARIANT(%s) AS %s) AS %s',
+                            SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                            $destinationColumn->getColumnDefinition()->getSQLDefinition(),
+                            SnowflakeQuote::quoteSingleIdentifier($destinationColumn->getColumnName())
+                        );
+                        continue;
+                    }
+                    $columnsSetSql[] = sprintf(
+                        'CAST(%s AS %s) AS %s',
+                        SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                        $destinationColumn->getColumnDefinition()->getSQLDefinition(),
+                        SnowflakeQuote::quoteSingleIdentifier($destinationColumn->getColumnName())
+                    );
+                    continue;
+                }
+                $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName());
                 continue;
             }
 
             // Input mapping convert empty values to null
             // empty strings '' are converted to null values
-            if (in_array($columnDefinition->getColumnName(), $importOptions->getConvertEmptyValuesToNull(), true)) {
+            if (in_array($sourceColumn->getColumnName(), $importOptions->getConvertEmptyValuesToNull(), true)) {
                 // use nullif only for string base type
-                if ($columnDefinition->getColumnDefinition()->getBasetype() === BaseType::STRING) {
+                if ($sourceColumn->getColumnDefinition()->getBasetype() === BaseType::STRING) {
                     $columnsSetSql[] = sprintf(
                         'IFF(%s = \'\', NULL, %s)',
-                        SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName()),
-                        SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName())
+                        SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                        SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName())
                     );
                     continue;
                 }
                 // if tables is not typed column could be other than string in this case we skip conversion
-                $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName());
+                $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName());
                 continue;
             }
 
             // for string base type convert null values to empty string ''
             //phpcs:ignore
-            if (!$importOptions->usingUserDefinedTypes() && $columnDefinition->getColumnDefinition()->getBasetype() === BaseType::STRING) {
+            if (!$importOptions->usingUserDefinedTypes() && $sourceColumn->getColumnDefinition()->getBasetype() === BaseType::STRING) {
                 $columnsSetSql[] = sprintf(
                     'COALESCE(%s, \'\') AS %s',
-                    SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName()),
-                    SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName())
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName())
                 );
                 continue;
             }
             // on columns other than string dont use COALESCE
             // this will fail if the column is not null, but this is expected
-            $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($columnDefinition->getColumnName());
+            $columnsSetSql[] = SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName());
         }
 
         if ($useTimestamp) {
@@ -256,29 +281,58 @@ public function getUpdateWithPkCommand(
         SnowflakeImportOptions $importOptions,
         string $timestamp
     ): string {
+        $columnMap = SourceDestinationColumnMap::createForTables(
+            $stagingTableDefinition,
+            $destinationDefinition,
+            $importOptions->ignoreColumns()
+        );
         $columnsSet = [];
 
-        foreach ($stagingTableDefinition->getColumnsNames() as $columnName) {
+        foreach ($stagingTableDefinition->getColumnsDefinitions() as $sourceColumn) {
             if (!$importOptions->isNullManipulationEnabled()) {
+                $destinationColumn = $columnMap->getDestination($sourceColumn);
+                $type = $destinationColumn->getColumnDefinition()->getType();
+                $useAutoCast = in_array($type, $importOptions->autoCastTypes(), true);
+                $isSameType = $type === $sourceColumn->getColumnDefinition()->getType();
+                if ($useAutoCast && !$isSameType) {
+                    if ($type === Snowflake::TYPE_OBJECT) {
+                        // object can't be casted from string but can be casted from variant
+                        $columnsSet[] = sprintf(
+                            '%s = CAST(TO_VARIANT("src".%s) AS %s)',
+                            SnowflakeQuote::quoteSingleIdentifier($destinationColumn->getColumnName()),
+                            SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                            $destinationColumn->getColumnDefinition()->getSQLDefinition(),
+                        );
+                        continue;
+                    }
+                    $columnsSet[] = sprintf(
+                        '%s = CAST("src".%s AS %s)',
+                        SnowflakeQuote::quoteSingleIdentifier($destinationColumn->getColumnName()),
+                        SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                        $destinationColumn->getColumnDefinition()->getSQLDefinition(),
+                    );
+                    continue;
+                }
+
                 $columnsSet[] = sprintf(
                     '%s = "src".%s',
-                    SnowflakeQuote::quoteSingleIdentifier($columnName),
-                    SnowflakeQuote::quoteSingleIdentifier($columnName),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
                 );
                 continue;
             }
-            if (in_array($columnName, $importOptions->getConvertEmptyValuesToNull(), true)) {
+            if (in_array($sourceColumn->getColumnName(), $importOptions->getConvertEmptyValuesToNull(), true)) {
                 $columnsSet[] = sprintf(
                     '%s = IFF("src".%s = \'\', NULL, "src".%s)',
-                    SnowflakeQuote::quoteSingleIdentifier($columnName),
-                    SnowflakeQuote::quoteSingleIdentifier($columnName),
-                    SnowflakeQuote::quoteSingleIdentifier($columnName)
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName())
                 );
             } else {
                 $columnsSet[] = sprintf(
                     '%s = COALESCE("src".%s, \'\')',
-                    SnowflakeQuote::quoteSingleIdentifier($columnName),
-                    SnowflakeQuote::quoteSingleIdentifier($columnName)
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName()),
+                    SnowflakeQuote::quoteSingleIdentifier($sourceColumn->getColumnName())
                 );
             }
         }
diff --git a/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
new file mode 100644
index 000000000..0cbfdba2e
--- /dev/null
+++ b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
@@ -0,0 +1,91 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Keboola\Db\ImportExport\Backend;
+
+use Exception;
+use Keboola\Db\ImportExport\Exception\ColumnsMismatchException;
+use Keboola\TableBackendUtils\Column\ColumnCollection;
+use Keboola\TableBackendUtils\Column\ColumnInterface;
+use Keboola\TableBackendUtils\Table\TableDefinitionInterface;
+use WeakMap;
+
+/**
+ * Class will create map of table column based on columns order
+ */
+final class SourceDestinationColumnMap
+{
+    /**
+     * @var WeakMap<ColumnInterface,ColumnInterface>
+     */
+    private WeakMap $map;
+
+    /**
+     * @param string[] $ignoreColumns
+     */
+    public function __construct(
+        private readonly ColumnCollection $source,
+        private readonly ColumnCollection $destination,
+        private readonly array $ignoreColumns = [],
+    ) {
+        $this->map = new WeakMap();
+        $this->buildMap();
+    }
+
+    /**
+     * @param string[] $ignoreColumns
+     */
+    public static function createForTables(
+        TableDefinitionInterface $source,
+        TableDefinitionInterface $destination,
+        array $ignoreColumns = [],
+    ): self {
+        return new self(
+            $source->getColumnsDefinitions(),
+            $destination->getColumnsDefinitions(),
+            $ignoreColumns
+        );
+    }
+
+    private function buildMap(): void
+    {
+        $it0 = $this->source->getIterator();
+        $it1 = $this->destination->getIterator();
+        while ($it0->valid() || $it1->valid()) {
+            if ($it0->valid() && in_array($it0->current()->getColumnName(), $this->ignoreColumns, true)) {
+                $it0->next();
+                if (!$it0->valid() && !$it1->valid()) {
+                    break;
+                }
+            }
+            if ($it1->valid() && in_array($it1->current()->getColumnName(), $this->ignoreColumns, true)) {
+                $it1->next();
+                if (!$it0->valid() && !$it1->valid()) {
+                    break;
+                }
+            }
+            if ($it0->valid() && $it1->valid()) {
+                /** @var ColumnInterface $sourceCol */
+                $sourceCol = $it0->current();
+                /** @var ColumnInterface $destCol */
+                $destCol = $it1->current();
+                $this->map[$sourceCol] = $destCol;
+            } else {
+                throw ColumnsMismatchException::createColumnsCountMismatch($this->source, $this->destination);
+            }
+            $it0->next();
+            $it1->next();
+        }
+    }
+
+    public function getDestination(ColumnInterface $source): ColumnInterface
+    {
+        $destination = $this->map[$source];
+        if (!$destination instanceof ColumnInterface) {
+            // this can happen only when class is used with different source and destination tables instances
+            throw new Exception(sprintf('Column "%s" not found in destination table', $source->getColumnName()));
+        }
+        return $destination;
+    }
+}
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
index 289b457ff..b75ac380e 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
@@ -7,6 +7,8 @@
 use Generator;
 use Keboola\Csv\CsvFile;
 use Keboola\CsvOptions\CsvOptions;
+use Keboola\Datatype\Definition\Snowflake;
+use Keboola\Db\ImportExport\Backend\ImportState;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\FullImporter;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\SqlBuilder;
@@ -33,6 +35,97 @@ protected function setUp(): void
         $this->createSchema($this->getDestinationSchemaName());
     }
 
+    /**
+     * Test is testing loading of semi-structured data into typed table.
+     *
+     * We ignore here GEOGRAPHY and GEOMETRY as they act differently when casting from string
+     * https://docs.snowflake.com/en/sql-reference/functions/to_geography
+     * https://docs.snowflake.com/en/sql-reference/functions/to_geometry
+     *
+     * This test is not using CSV but inserting data directly into stage table to mimic this behavior
+     */
+    public function testLoadTypedTableWithCastingValues(): void
+    {
+        $this->connection->executeQuery(sprintf(
+            /** @lang Snowflake */
+            'CREATE TABLE %s."types" (
+              "id"  NUMBER,
+              "VARIANT" VARIANT,
+              "BINARY" BINARY,
+              "VARBINARY" VARBINARY,
+              "OBJECT" OBJECT,
+              "ARRAY" ARRAY,
+              "_timestamp" TIMESTAMP
+            );',
+            SnowflakeQuote::quoteSingleIdentifier($this->getDestinationSchemaName())
+        ));
+
+        // skipping header
+        $options = new SnowflakeImportOptions(
+            [],
+            false,
+            false,
+            1,
+            SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
+            SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            ['_timestamp'],
+            [
+                Snowflake::TYPE_VARIANT,
+                Snowflake::TYPE_BINARY,
+                Snowflake::TYPE_VARBINARY,
+                Snowflake::TYPE_OBJECT,
+                Snowflake::TYPE_ARRAY,
+            ]
+        );
+
+        $destinationRef = new SnowflakeTableReflection(
+            $this->connection,
+            $this->getDestinationSchemaName(),
+            'types'
+        );
+        /** @var SnowflakeTableDefinition $destination */
+        $destination = $destinationRef->getTableDefinition();
+        $stagingTable = StageTableDefinitionFactory::createVarcharStagingTableDefinition(
+            $destination->getSchemaName(),
+            [
+                'id',
+                'VARIANT',
+                'BINARY',
+                'VARBINARY',
+                'OBJECT',
+                'ARRAY',
+            ]
+        );
+
+        $qb = new SnowflakeTableQueryBuilder();
+        $this->connection->executeStatement(
+            $qb->getCreateTableCommandFromDefinition($stagingTable)
+        );
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+select 1, 
+       TO_VARCHAR(TO_VARIANT(\'3.14\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+;',
+            $stagingTable->getSchemaName(),
+            $stagingTable->getTableName()
+        ));
+        $toFinalTableImporter = new FullImporter($this->connection);
+
+        $toFinalTableImporter->importToTable(
+            $stagingTable,
+            $destination,
+            $options,
+            new ImportState($stagingTable->getTableName())
+        );
+
+        self::assertEquals(1, $destinationRef->getRowsCount());
+    }
+
     public function testLoadToTableWithNullValuesShouldPass(): void
     {
         $this->initTable(self::TABLE_SINGLE_PK);
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
index 1ebf978e9..fae6c362d 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
@@ -6,6 +6,8 @@
 
 use Generator;
 use Keboola\Csv\CsvFile;
+use Keboola\Datatype\Definition\Snowflake;
+use Keboola\Db\ImportExport\Backend\ImportState;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\FullImporter;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\IncrementalImporter;
@@ -14,6 +16,7 @@
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\ToStageImporter;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Storage;
+use Keboola\TableBackendUtils\Escaping\Snowflake\SnowflakeQuote;
 use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableDefinition;
 use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableQueryBuilder;
 use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableReflection;
@@ -42,6 +45,124 @@ protected function setUp(): void
         $this->createSchema($this->getDestinationSchemaName());
     }
 
+    /**
+     * Test is testing loading of semi-structured data into typed table.
+     *
+     * We ignore here GEOGRAPHY and GEOMETRY as they act differently when casting from string
+     * https://docs.snowflake.com/en/sql-reference/functions/to_geography
+     * https://docs.snowflake.com/en/sql-reference/functions/to_geometry
+     *
+     * This test is not using CSV but inserting data directly into stage table to mimic this behavior
+     */
+    public function testLoadTypedTableWithCastingValues(): void
+    {
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'CREATE TABLE %s."types" (
+              "id" NUMBER,
+              "VARIANT" VARIANT,
+              "BINARY" BINARY,
+              "VARBINARY" VARBINARY,
+              "OBJECT" OBJECT,
+              "ARRAY" ARRAY,
+              "_timestamp" TIMESTAMP,
+               PRIMARY KEY ("id")
+            );',
+            SnowflakeQuote::quoteSingleIdentifier($this->getDestinationSchemaName())
+        ));
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT 1, 
+      TO_VARIANT(\'3.14\'),
+      TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+      TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+      OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT),
+      ARRAY_CONSTRUCT(1, 2, 3, NULL)
+;',
+            $this->getDestinationSchemaName(),
+            'types'
+        ));
+
+        // skipping header
+        $options = new SnowflakeImportOptions(
+            [],
+            false,
+            false,
+            1,
+            SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
+            SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            ['_timestamp'],
+            [
+                Snowflake::TYPE_VARIANT,
+                Snowflake::TYPE_BINARY,
+                Snowflake::TYPE_VARBINARY,
+                Snowflake::TYPE_OBJECT,
+                Snowflake::TYPE_ARRAY,
+            ]
+        );
+
+        $destinationRef = new SnowflakeTableReflection(
+            $this->connection,
+            $this->getDestinationSchemaName(),
+            'types'
+        );
+        /** @var SnowflakeTableDefinition $destination */
+        $destination = $destinationRef->getTableDefinition();
+        $stagingTable = StageTableDefinitionFactory::createVarcharStagingTableDefinition(
+            $destination->getSchemaName(),
+            [
+                'id',
+                'VARIANT',
+                'BINARY',
+                'VARBINARY',
+                'OBJECT',
+                'ARRAY',
+            ]
+        );
+
+        $qb = new SnowflakeTableQueryBuilder();
+        $this->connection->executeStatement(
+            $qb->getCreateTableCommandFromDefinition($stagingTable)
+        );
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT 1, 
+       TO_VARCHAR(TO_VARIANT(\'3.14\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+;',
+            $stagingTable->getSchemaName(),
+            $stagingTable->getTableName()
+        ));
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT 2, 
+       TO_VARCHAR(TO_VARIANT(\'3.14\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+;',
+            $stagingTable->getSchemaName(),
+            $stagingTable->getTableName()
+        ));
+        $toFinalTableImporter = new IncrementalImporter($this->connection);
+
+        $toFinalTableImporter->importToTable(
+            $stagingTable,
+            $destination,
+            $options,
+            new ImportState($stagingTable->getTableName())
+        );
+
+        self::assertEquals(2, $destinationRef->getRowsCount());
+    }
+
     /**
      * @return \Generator<string, array<mixed>>
      */

From 4c4efe018fb6ae814e8dce3004a8b49da5469dc7 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Tue, 19 Sep 2023 20:14:40 +0200
Subject: [PATCH 02/10] ignore _timestamp column in tests

---
 .../Snowflake/SnowflakeBaseTestCase.php       | 19 ++++---
 .../Snowflake/ToFinal/FullImportTest.php      | 12 +++--
 .../ToFinal/IncrementalImportTest.php         | 53 ++++++++++---------
 .../Snowflake/ToFinal/SqlBuilderTest.php      | 53 ++++++++++---------
 .../Snowflake/ToStage/StageImportTest.php     | 34 ++++++------
 5 files changed, 95 insertions(+), 76 deletions(-)

diff --git a/packages/php-db-import-export/tests/functional/Snowflake/SnowflakeBaseTestCase.php b/packages/php-db-import-export/tests/functional/Snowflake/SnowflakeBaseTestCase.php
index d4588d2a5..166b7fa5b 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/SnowflakeBaseTestCase.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/SnowflakeBaseTestCase.php
@@ -9,6 +9,7 @@
 use Doctrine\DBAL\Logging\Middleware;
 use Exception;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Storage\SourceInterface;
 use Keboola\TableBackendUtils\Connection\Snowflake\SnowflakeConnectionFactory;
@@ -403,10 +404,11 @@ protected function getSnowflakeImportOptions(
         bool $useTimeStamp = true
     ): SnowflakeImportOptions {
         return new SnowflakeImportOptions(
-            [],
-            false,
-            $useTimeStamp,
-            $skipLines
+            convertEmptyValuesToNull: [],
+            isIncremental: false,
+            useTimestamp: $useTimeStamp,
+            numberOfIgnoredLines: $skipLines,
+            ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME]
         );
     }
 
@@ -472,10 +474,11 @@ protected function getSimpleImportOptions(
         int $skipLines = ImportOptions::SKIP_FIRST_LINE
     ): SnowflakeImportOptions {
         return new SnowflakeImportOptions(
-            [],
-            false,
-            true,
-            $skipLines
+            convertEmptyValuesToNull: [],
+            isIncremental: false,
+            useTimestamp: true,
+            numberOfIgnoredLines: $skipLines,
+            ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
         );
     }
 }
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
index b75ac380e..acd23d720 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
@@ -14,6 +14,7 @@
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\SqlBuilder;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\StageTableDefinitionFactory;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\ToStageImporter;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Storage\Snowflake\Table;
 use Keboola\Db\ImportExport\Storage\SourceInterface;
@@ -68,7 +69,7 @@ public function testLoadTypedTableWithCastingValues(): void
             1,
             SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
             SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
-            ['_timestamp'],
+            [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             [
                 Snowflake::TYPE_VARIANT,
                 Snowflake::TYPE_BINARY,
@@ -603,10 +604,11 @@ public function fullImportData(): Generator
                 self::TABLE_OUT_NO_TIMESTAMP_TABLE,
             ],
             new SnowflakeImportOptions(
-                [],
-                false,
-                false, // don't use timestamp
-                ImportOptions::SKIP_FIRST_LINE
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false, // don't use timestamp
+                numberOfIgnoredLines: ImportOptions::SKIP_FIRST_LINE,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             $escapingStub->getRows(),
             7,
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
index fae6c362d..1d7002d3b 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
@@ -14,6 +14,7 @@
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\SqlBuilder;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\StageTableDefinitionFactory;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\ToStageImporter;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\Db\ImportExport\Storage;
 use Keboola\TableBackendUtils\Escaping\Snowflake\SnowflakeQuote;
@@ -29,10 +30,11 @@ protected function getSnowflakeIncrementalImportOptions(
         int $skipLines = ImportOptions::SKIP_FIRST_LINE
     ): SnowflakeImportOptions {
         return new SnowflakeImportOptions(
-            [],
-            true,
-            true,
-            $skipLines
+            convertEmptyValuesToNull: [],
+            isIncremental: true,
+            useTimestamp: true,
+            numberOfIgnoredLines: $skipLines,
+            ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
         );
     }
 
@@ -86,14 +88,14 @@ public function testLoadTypedTableWithCastingValues(): void
 
         // skipping header
         $options = new SnowflakeImportOptions(
-            [],
-            false,
-            false,
-            1,
-            SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
-            SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
-            ['_timestamp'],
-            [
+            convertEmptyValuesToNull: [],
+            isIncremental: false,
+            useTimestamp: false,
+            numberOfIgnoredLines: 1,
+            requireSameTables: SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
+            nullManipulation: SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
+            autoCastTypes: [
                 Snowflake::TYPE_VARIANT,
                 Snowflake::TYPE_BINARY,
                 Snowflake::TYPE_VARBINARY,
@@ -204,10 +206,11 @@ public function incrementalImportData(): Generator
                 ['id']
             ),
             new SnowflakeImportOptions(
-                [],
-                false,
-                false, // disable timestamp
-                ImportOptions::SKIP_FIRST_LINE
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false, // disable timestamp
+                numberOfIgnoredLines: ImportOptions::SKIP_FIRST_LINE,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             $this->getSourceInstance(
                 'tw_accounts.increment.csv',
@@ -217,10 +220,11 @@ public function incrementalImportData(): Generator
                 ['id']
             ),
             new SnowflakeImportOptions(
-                [],
-                true, // incremental
-                false, // disable timestamp
-                ImportOptions::SKIP_FIRST_LINE
+                convertEmptyValuesToNull: [],
+                isIncremental: true, // incremental
+                useTimestamp: false, // disable timestamp
+                numberOfIgnoredLines: ImportOptions::SKIP_FIRST_LINE,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             [$this->getDestinationSchemaName(), 'accounts_without_ts'],
             $accountsStub->getRows(),
@@ -259,10 +263,11 @@ public function incrementalImportData(): Generator
                 ['VisitID', 'Value', 'MenuItem']
             ),
             new SnowflakeImportOptions(
-                [],
-                true, // incremental
-                false, // disable timestamp
-                ImportOptions::SKIP_FIRST_LINE
+                convertEmptyValuesToNull: [],
+                isIncremental: true, // incremental
+                useTimestamp: false, // disable timestamp
+                numberOfIgnoredLines: ImportOptions::SKIP_FIRST_LINE,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             $this->getSourceInstance(
                 'multi-pk.increment.csv',
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index 05f052135..c859c3a9a 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -9,6 +9,7 @@
 use Keboola\Db\ImportExport\Backend\Snowflake\Helper\DateTimeHelper;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\SqlBuilder;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
 use Keboola\Db\ImportExport\ImportOptions;
 use Keboola\TableBackendUtils\Column\ColumnCollection;
 use Keboola\TableBackendUtils\Column\Snowflake\SnowflakeColumn;
@@ -303,12 +304,13 @@ public function testGetDeleteOldItemsCommandRequireSameTables(): void
             $stagingTableDefinition,
             $tableDefinition,
             new SnowflakeImportOptions(
-                [],
-                false,
-                false,
-                0,
-                ImportOptions::SAME_TABLES_NOT_REQUIRED,
-                ImportOptions::NULL_MANIPULATION_SKIP //<- skipp null manipulation
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false,
+                numberOfIgnoredLines: 0,
+                requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
+                nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
         );
 
@@ -475,12 +477,13 @@ public function testGetInsertAllIntoTargetTableCommandSameTables(): void
             $fakeStage,
             $destination,
             new SnowflakeImportOptions(
-                [],
-                false,
-                false,
-                0,
-                ImportOptions::SAME_TABLES_NOT_REQUIRED,
-                ImportOptions::NULL_MANIPULATION_SKIP //<- skipp null manipulation
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false,
+                numberOfIgnoredLines: 0,
+                requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
+                nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             '2020-01-01 00:00:00'
         );
@@ -832,12 +835,13 @@ public function testGetUpdateWithPkCommandRequireSameTables(): void
             $fakeStage,
             $fakeDestination,
             new SnowflakeImportOptions(
-                [],
-                false,
-                false,
-                0,
-                ImportOptions::SAME_TABLES_NOT_REQUIRED,
-                ImportOptions::NULL_MANIPULATION_SKIP //<- skipp null manipulation
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false,
+                numberOfIgnoredLines: 0,
+                requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
+                nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             ),
             '2020-01-01 00:00:00'
         );
@@ -1110,12 +1114,13 @@ public function testGetUpdateWithPkCommandNullManipulationWithTimestamp(): void
 
         // use timestamp
         $options = new SnowflakeImportOptions(
-            ['col1'],
-            false,
-            true,
-            0,
-            SnowflakeImportOptions::SAME_TABLES_REQUIRED,
-            SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            convertEmptyValuesToNull: ['col1'],
+            isIncremental: false,
+            useTimestamp: true,
+            numberOfIgnoredLines: 0,
+            requireSameTables: SnowflakeImportOptions::SAME_TABLES_REQUIRED,
+            nullManipulation: SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
         );
         $sql = $this->getBuilder()->getUpdateWithPkCommand(
             $fakeStage,
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToStage/StageImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToStage/StageImportTest.php
index 50459d1b2..a3abff752 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToStage/StageImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToStage/StageImportTest.php
@@ -7,6 +7,7 @@
 use Keboola\Db\Import\Exception;
 use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
 use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\ToStageImporter;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
 use Keboola\Db\ImportExport\Exception\ColumnsMismatchException;
 use Keboola\Db\ImportExport\Storage\Snowflake\Table;
 use Keboola\TableBackendUtils\Escaping\Snowflake\SnowflakeQuote;
@@ -176,11 +177,12 @@ public function testMoveDataFromAToBRequireSameTablesFailColumnNameMismatch(): v
             $source,
             $targetTableRef->getTableDefinition(),
             new SnowflakeImportOptions(
-                [],
-                false,
-                true,
-                1,
-                SnowflakeImportOptions::SAME_TABLES_REQUIRED
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: true,
+                numberOfIgnoredLines: 1,
+                requireSameTables: SnowflakeImportOptions::SAME_TABLES_REQUIRED,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             )
         );
     }
@@ -225,11 +227,12 @@ public function testMoveDataFromAToBRequireSameTablesFailColumnCount(): void
             $source,
             $targetTableRef->getTableDefinition(),
             new SnowflakeImportOptions(
-                [],
-                false,
-                true,
-                1,
-                SnowflakeImportOptions::SAME_TABLES_REQUIRED
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: true,
+                numberOfIgnoredLines: 1,
+                requireSameTables: SnowflakeImportOptions::SAME_TABLES_REQUIRED,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             )
         );
     }
@@ -272,11 +275,12 @@ public function testMoveDataFromAToBRequireSameTablesFailColumnMismatch(): void
             $source,
             $targetTableRef->getTableDefinition(),
             new SnowflakeImportOptions(
-                [],
-                false,
-                true,
-                1,
-                SnowflakeImportOptions::SAME_TABLES_REQUIRED
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: true,
+                numberOfIgnoredLines: 1,
+                requireSameTables: SnowflakeImportOptions::SAME_TABLES_REQUIRED,
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
             )
         );
     }

From 099a759494a7fd532680fb3441b0d301f80cee6d Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Tue, 19 Sep 2023 20:34:18 +0200
Subject: [PATCH 03/10] add SourceDestinationColumnMapTest

---
 .../Backend/SourceDestinationColumnMap.php    |   7 +-
 .../SourceDestinationColumnMapTest.php        | 210 ++++++++++++++++++
 2 files changed, 215 insertions(+), 2 deletions(-)
 create mode 100644 packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php

diff --git a/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
index 0cbfdba2e..798e2a167 100644
--- a/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
+++ b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
@@ -4,6 +4,7 @@
 
 namespace Keboola\Db\ImportExport\Backend;
 
+use Error;
 use Exception;
 use Keboola\Db\ImportExport\Exception\ColumnsMismatchException;
 use Keboola\TableBackendUtils\Column\ColumnCollection;
@@ -81,11 +82,13 @@ private function buildMap(): void
 
     public function getDestination(ColumnInterface $source): ColumnInterface
     {
-        $destination = $this->map[$source];
-        if (!$destination instanceof ColumnInterface) {
+        try {
+            $destination = $this->map[$source];
+        } catch (Error $e) {
             // this can happen only when class is used with different source and destination tables instances
             throw new Exception(sprintf('Column "%s" not found in destination table', $source->getColumnName()));
         }
+        assert($destination !== null);
         return $destination;
     }
 }
diff --git a/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
new file mode 100644
index 000000000..41b9d0dfc
--- /dev/null
+++ b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
@@ -0,0 +1,210 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Keboola\Db\ImportExportUnit\Backend;
+
+use Exception;
+use Keboola\Datatype\Definition\DefinitionInterface;
+use Keboola\Db\ImportExport\Backend\SourceDestinationColumnMap;
+use Keboola\Db\ImportExport\Exception\ColumnsMismatchException;
+use Keboola\TableBackendUtils\Column\ColumnCollection;
+use Keboola\TableBackendUtils\Column\ColumnInterface;
+use Keboola\TableBackendUtils\Table\TableDefinitionInterface;
+use PHPUnit\Framework\TestCase;
+use Throwable;
+
+/**
+ * Class SourceDestinationColumnMapTest.
+ *
+ * @covers \Keboola\Db\ImportExport\Backend\SourceDestinationColumnMap
+ */
+final class SourceDestinationColumnMapTest extends TestCase
+{
+    /**
+     * @return array{ColumnInterface, ColumnInterface, SourceDestinationColumnMap}
+     */
+    private function getMap(): array
+    {
+        $col1 = $this->getColumn('col1', 'string');
+        $source = new ColumnCollection([
+            $col1,
+            $this->getColumn('col2', 'string'),
+        ]);
+        $col1Dest = $this->getColumn('col1', 'bool');
+        $destination = new ColumnCollection([
+            $col1Dest,
+            $this->getColumn('col2', 'bool'),
+        ]);
+
+        $map = new SourceDestinationColumnMap(
+            $source,
+            $destination
+        );
+        return [$col1, $col1Dest, $map];
+    }
+
+    private function getColumn(string $name, string $type): ColumnInterface
+    {
+        return new class($name, $type) implements ColumnInterface {
+            public function __construct(private readonly string $name, private readonly string $type)
+            {
+            }
+
+            public function getColumnName(): string
+            {
+                return $this->name;
+            }
+
+            public function getColumnDefinition(): DefinitionInterface
+            {
+                return new class($this->type) implements DefinitionInterface {
+                    public function __construct(private readonly string $type)
+                    {
+                    }
+
+                    public function getSQLDefinition(): string
+                    {
+                        return $this->type . 'DEF';
+                    }
+
+                    public function toArray(): array
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public function getBasetype(): string
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public function getType(): string
+                    {
+                        return $this->type;
+                    }
+
+                    public function getLength(): ?string
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public function isNullable(): bool
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public function getDefault(): ?string
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public static function getTypeByBasetype(string $basetype): string
+                    {
+                        throw new Exception('Not implemented');
+                    }
+
+                    public function isSameType(DefinitionInterface $definition): bool
+                    {
+                        throw new Exception('Not implemented');
+                    }
+                };
+            }
+
+            public static function createGenericColumn(string $columnName): ColumnInterface
+            {
+                throw new Exception('Not implemented');
+            }
+
+            public static function createTimestampColumn(
+                string $columnName = self::TIMESTAMP_COLUMN_NAME
+            ): ColumnInterface {
+                throw new Exception('Not implemented');
+            }
+
+            /**
+             * @param array<mixed> $dbResponse
+             */
+            public static function createFromDB(array $dbResponse): ColumnInterface
+            {
+                throw new Exception('Not implemented');
+            }
+        };
+    }
+
+    public function testCreateForTables(): void
+    {
+        $col1 = $this->getColumn('col1', 'string');
+        $source = $this->createMock(TableDefinitionInterface::class);
+        $source->expects(self::once())->method('getColumnsDefinitions')->willReturn(new ColumnCollection([
+            $col1,
+            $this->getColumn('col2', 'string'),
+        ]));
+        $col1Dest = $this->getColumn('col1', 'bool');
+        $destination = $this->createMock(TableDefinitionInterface::class);
+        $destination->expects(self::once())->method('getColumnsDefinitions')->willReturn(new ColumnCollection([
+            $col1Dest,
+            $this->getColumn('col2', 'bool'),
+        ]));
+
+        $map = SourceDestinationColumnMap::createForTables(
+            $source,
+            $destination
+        );
+
+        $this->assertSame($col1Dest, $map->getDestination($col1));
+    }
+
+    public function testCreateForCollection(): void
+    {
+        [$col1, $col1Dest, $map] = $this->getMap();
+
+        $this->assertSame($col1Dest, $map->getDestination($col1));
+    }
+
+    public function testColumnMismatch(): void
+    {
+        $source = new ColumnCollection([
+            $this->getColumn('col1', 'string'),
+            $this->getColumn('col2', 'string'),
+        ]);
+        $destination = new ColumnCollection([
+            $this->getColumn('col1', 'bool'),
+            $this->getColumn('col2', 'bool'),
+            $this->getColumn('col3', 'bool'),
+        ]);
+
+        $this->expectException(ColumnsMismatchException::class);
+        new SourceDestinationColumnMap(
+            $source,
+            $destination
+        );
+    }
+
+    public function testIgnoreColumn(): void
+    {
+        $this->expectNotToPerformAssertions();
+        $source = new ColumnCollection([
+            $this->getColumn('col1', 'string'),
+            $this->getColumn('col2', 'string'),
+        ]);
+        $destination = new ColumnCollection([
+            $this->getColumn('col1', 'bool'),
+            $this->getColumn('col2', 'bool'),
+            $this->getColumn('col3', 'bool'),
+        ]);
+
+        new SourceDestinationColumnMap(
+            $source,
+            $destination,
+            ['col3']
+        );
+    }
+
+    public function testColumnNotFound(): void
+    {
+        [, , $map] = $this->getMap();
+
+        $this->expectException(Throwable::class);
+        $map->getDestination($this->getColumn('test', 'string'));
+    }
+}

From 66953f21b0ca12111909f488aff70b76e0e37e92 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 20 Sep 2023 07:58:55 +0200
Subject: [PATCH 04/10] fix test cases

---
 .../Snowflake/ToFinal/FullImportTest.php      | 11 +++++++++-
 .../Snowflake/ToFinal/SqlBuilderTest.php      | 20 +++++++++++++++----
 2 files changed, 26 insertions(+), 5 deletions(-)

diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
index acd23d720..077f482ef 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
@@ -560,7 +560,16 @@ public function fullImportData(): Generator
                 []
             ),
             [$this->getDestinationSchemaName(), self::TABLE_TABLE],
-            $this->getSnowflakeImportOptions(),
+            new SnowflakeImportOptions(
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: true,
+                numberOfIgnoredLines: 1,
+                ignoreColumns: [
+                    ToStageImporterInterface::TIMESTAMP_COLUMN_NAME,
+                    'lemmaIndex'
+                ]
+            ),
             [['table', 'column', null]],
             1,
             self::TABLE_TABLE,
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index c859c3a9a..98e4cd60b 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -412,7 +412,9 @@ public function testGetInsertAllIntoTargetTableCommand(): void
         $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
             $fakeStage,
             $destination,
-            $this->getDummyImportOptions(),
+            new SnowflakeImportOptions(
+                ignoreColumns: ['id']
+            ),
             '2020-01-01 00:00:00'
         );
 
@@ -483,7 +485,9 @@ public function testGetInsertAllIntoTargetTableCommandSameTables(): void
                 numberOfIgnoredLines: 0,
                 requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
                 nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
-                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
+                ignoreColumns: [
+                    'id',
+                ],
             ),
             '2020-01-01 00:00:00'
         );
@@ -599,7 +603,10 @@ public function testGetInsertAllIntoTargetTableCommandConvertToNull(): void
         );
 
         // convert col1 to null
-        $options = new SnowflakeImportOptions(['col1']);
+        $options = new SnowflakeImportOptions(
+            convertEmptyValuesToNull: ['col1'],
+            ignoreColumns: ['id'],
+        );
         $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
             $fakeStage,
             $destination,
@@ -661,7 +668,12 @@ public function testGetInsertAllIntoTargetTableCommandConvertToNullWithTimestamp
         );
 
         // use timestamp
-        $options = new SnowflakeImportOptions(['col1'], false, true);
+        $options = new SnowflakeImportOptions(
+            convertEmptyValuesToNull: ['col1'],
+            isIncremental: false,
+            useTimestamp: true,
+            ignoreColumns: ['id'],
+        );
         $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
             $fakeStage,
             $destination,

From e69b473996a742f20a1d8cd192a1623e1126a5dc Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 20 Sep 2023 09:14:33 +0200
Subject: [PATCH 05/10] add functional tests for sql builder

---
 .../Snowflake/ToFinal/FullImportTest.php      |   2 +-
 .../Snowflake/ToFinal/SqlBuilderTest.php      | 316 ++++++++++++++++++
 2 files changed, 317 insertions(+), 1 deletion(-)

diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
index 077f482ef..be2443455 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
@@ -567,7 +567,7 @@ public function fullImportData(): Generator
                 numberOfIgnoredLines: 1,
                 ignoreColumns: [
                     ToStageImporterInterface::TIMESTAMP_COLUMN_NAME,
-                    'lemmaIndex'
+                    'lemmaIndex',
                 ]
             ),
             [['table', 'column', null]],
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index 98e4cd60b..22e138c65 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -456,6 +456,175 @@ public function testGetInsertAllIntoTargetTableCommand(): void
         ], $result);
     }
 
+    public function testGetInsertAllIntoTargetTableCommandCasting(): void
+    {
+        $this->createTestSchema();
+        $destination = new SnowflakeTableDefinition(
+            self::TEST_SCHEMA,
+            self::TEST_TABLE,
+            false,
+            new ColumnCollection([
+                $this->createNullableGenericColumn('pk1'),
+                new SnowflakeColumn(
+                    'VARIANT',
+                    new Snowflake(
+                        Snowflake::TYPE_VARIANT
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'BINARY',
+                    new Snowflake(
+                        Snowflake::TYPE_BINARY
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'VARBINARY',
+                    new Snowflake(
+                        Snowflake::TYPE_VARBINARY
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'OBJECT',
+                    new Snowflake(
+                        Snowflake::TYPE_OBJECT
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'ARRAY',
+                    new Snowflake(
+                        Snowflake::TYPE_ARRAY
+                    ),
+                ),
+            ]),
+            ['pk1']
+        );
+        $stage = new SnowflakeTableDefinition(
+            self::TEST_SCHEMA,
+            self::TEST_STAGING_TABLE,
+            true,
+            new ColumnCollection([
+                $this->createNullableGenericColumn('pk1'),
+                $this->createNullableGenericColumn('VARIANT'),
+                $this->createNullableGenericColumn('BINARY'),
+                $this->createNullableGenericColumn('VARBINARY'),
+                $this->createNullableGenericColumn('OBJECT'),
+                $this->createNullableGenericColumn('ARRAY'),
+            ]),
+            []
+        );
+        $this->connection->executeStatement(
+            (new SnowflakeTableQueryBuilder())->getCreateTableCommandFromDefinition($destination)
+        );
+        $this->connection->executeStatement(
+            (new SnowflakeTableQueryBuilder())->getCreateTableCommandFromDefinition($stage)
+        );
+
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("pk1","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT \'1\', 
+       TO_VARIANT(\'4.14\'),
+       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+       OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  24::VARIANT),
+       ARRAY_CONSTRUCT(1, 2, 3, NULL)
+;',
+            self::TEST_SCHEMA,
+            self::TEST_TABLE,
+        ));
+
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("pk1","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT \'1\', 
+       TO_VARCHAR(TO_VARIANT(\'3.14\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+;',
+            self::TEST_SCHEMA,
+            self::TEST_STAGING_TABLE,
+        ));
+
+        // no convert values no timestamp
+        $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
+            $stage,
+            $destination,
+            new SnowflakeImportOptions(
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false,
+                numberOfIgnoredLines: 0,
+                requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
+                nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
+                autoCastTypes: [
+                    Snowflake::TYPE_VARIANT,
+                    Snowflake::TYPE_BINARY,
+                    Snowflake::TYPE_VARBINARY,
+                    Snowflake::TYPE_OBJECT,
+                    Snowflake::TYPE_ARRAY,
+                ],
+            ),
+            '2020-01-01 00:00:00'
+        );
+
+        self::assertEquals(
+        // phpcs:ignore
+            'INSERT INTO "import_export_test_schema"."import_export_test_test" ("pk1", "VARIANT", "BINARY", "VARBINARY", "OBJECT", "ARRAY") (SELECT "pk1",CAST("VARIANT" AS VARIANT) AS "VARIANT",CAST("BINARY" AS BINARY) AS "BINARY",CAST("VARBINARY" AS VARBINARY) AS "VARBINARY",CAST(TO_VARIANT("OBJECT") AS OBJECT) AS "OBJECT",CAST("ARRAY" AS ARRAY) AS "ARRAY" FROM "import_export_test_schema"."__temp_stagingTable" AS "src")',
+            $sql
+        );
+
+        $out = $this->connection->executeStatement($sql);
+        self::assertEquals(1, $out);
+
+        $result = $this->connection->fetchAllAssociative(sprintf(
+            'SELECT * FROM %s',
+            self::TEST_TABLE_IN_SCHEMA
+        ));
+
+        self::assertEqualsCanonicalizing([
+            [
+                'pk1' => '1',
+                'VARIANT' => '"3.14"',
+                'BINARY' => '1',
+                'VARBINARY' => '1',
+                'OBJECT' => <<<EOD
+{
+  "age": 42,
+  "name": "Jones"
+}
+EOD,
+                'ARRAY' => <<<EOD
+[
+  "[1,2,3,undefined]"
+]
+EOD,
+            ],
+            [
+                'pk1' => '1',
+                'VARIANT' => '"4.14"',
+                'BINARY' => '1',
+                'VARBINARY' => '1',
+                'OBJECT' => <<<EOD
+{
+  "age": 24,
+  "name": "Jones"
+}
+EOD,
+                'ARRAY' => <<<EOD
+[
+  1,
+  2,
+  3,
+  undefined
+]
+EOD,
+            ],
+        ], $result);
+    }
+
     public function testGetInsertAllIntoTargetTableCommandSameTables(): void
     {
         $this->createTestSchema();
@@ -878,6 +1047,152 @@ public function testGetUpdateWithPkCommandRequireSameTables(): void
         ], $result);
     }
 
+    public function testGetUpdateWithPkCommandCasting(): void
+    {
+        $this->createTestSchema();
+        $destination = new SnowflakeTableDefinition(
+            self::TEST_SCHEMA,
+            self::TEST_TABLE,
+            false,
+            new ColumnCollection([
+                $this->createNullableGenericColumn('pk1'),
+                new SnowflakeColumn(
+                    'VARIANT',
+                    new Snowflake(
+                        Snowflake::TYPE_VARIANT
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'BINARY',
+                    new Snowflake(
+                        Snowflake::TYPE_BINARY
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'VARBINARY',
+                    new Snowflake(
+                        Snowflake::TYPE_VARBINARY
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'OBJECT',
+                    new Snowflake(
+                        Snowflake::TYPE_OBJECT
+                    ),
+                ),
+                new SnowflakeColumn(
+                    'ARRAY',
+                    new Snowflake(
+                        Snowflake::TYPE_ARRAY
+                    ),
+                ),
+            ]),
+            ['pk1']
+        );
+        $stage = new SnowflakeTableDefinition(
+            self::TEST_SCHEMA,
+            self::TEST_STAGING_TABLE,
+            true,
+            new ColumnCollection([
+                $this->createNullableGenericColumn('pk1'),
+                $this->createNullableGenericColumn('VARIANT'),
+                $this->createNullableGenericColumn('BINARY'),
+                $this->createNullableGenericColumn('VARBINARY'),
+                $this->createNullableGenericColumn('OBJECT'),
+                $this->createNullableGenericColumn('ARRAY'),
+            ]),
+            []
+        );
+        $this->connection->executeStatement(
+            (new SnowflakeTableQueryBuilder())->getCreateTableCommandFromDefinition($destination)
+        );
+        $this->connection->executeStatement(
+            (new SnowflakeTableQueryBuilder())->getCreateTableCommandFromDefinition($stage)
+        );
+
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("pk1","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT \'1\', 
+       TO_VARIANT(\'4.14\'),
+       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
+       OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT),
+       ARRAY_CONSTRUCT(1, 2, 3, NULL)
+;',
+            self::TEST_SCHEMA,
+            self::TEST_TABLE,
+        ));
+
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("pk1","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+SELECT \'1\', 
+       TO_VARCHAR(TO_VARIANT(\'3.14\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
+       TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+;',
+            self::TEST_SCHEMA,
+            self::TEST_STAGING_TABLE,
+        ));
+
+        // no convert values no timestamp
+        $sql = $this->getBuilder()->getUpdateWithPkCommand(
+            $stage,
+            $destination,
+            new SnowflakeImportOptions(
+                convertEmptyValuesToNull: [],
+                isIncremental: false,
+                useTimestamp: false,
+                numberOfIgnoredLines: 0,
+                requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
+                nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
+                ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
+                autoCastTypes: [
+                    Snowflake::TYPE_VARIANT,
+                    Snowflake::TYPE_BINARY,
+                    Snowflake::TYPE_VARBINARY,
+                    Snowflake::TYPE_OBJECT,
+                    Snowflake::TYPE_ARRAY,
+                ],
+            ),
+            '2020-01-01 00:00:00'
+        );
+        self::assertEquals(
+        // phpcs:ignore
+            'UPDATE "import_export_test_schema"."import_export_test_test" AS "dest" SET "pk1" = "src"."pk1", "VARIANT" = CAST("src"."VARIANT" AS VARIANT), "BINARY" = CAST("src"."BINARY" AS BINARY), "VARBINARY" = CAST("src"."VARBINARY" AS VARBINARY), "OBJECT" = CAST(TO_VARIANT("src"."OBJECT") AS OBJECT), "ARRAY" = CAST("src"."ARRAY" AS ARRAY) FROM "import_export_test_schema"."__temp_stagingTable" AS "src" WHERE "dest"."pk1" = "src"."pk1" ',
+            $sql
+        );
+        $this->connection->executeStatement($sql);
+
+        $result = $this->connection->fetchAllAssociative(sprintf(
+            'SELECT * FROM %s',
+            self::TEST_TABLE_IN_SCHEMA
+        ));
+
+        self::assertEquals([
+            [
+                'pk1' => '1',
+                'VARIANT' => '"3.14"',
+                'BINARY' => '1',
+                'VARBINARY' => '1',
+                'OBJECT' => <<<EOD
+{
+  "age": 42,
+  "name": "Jones"
+}
+EOD,
+                'ARRAY' => <<<EOD
+[
+  "[1,2,3,undefined]"
+]
+EOD,
+            ],
+        ], $result);
+    }
+
     public function testGetUpdateWithPkCommandConvertValues(): void
     {
         $this->createTestSchema();
@@ -1071,6 +1386,7 @@ public function testGetUpdateWithPkCommandConvertValuesWithTimestamp(): void
             );
         }
     }
+
     public function testGetUpdateWithPkCommandNullManipulationWithTimestamp(): void
     {
         $timestampInit = new DateTime('2020-01-01 00:00:01');

From 0c0390f9fe888fc5a3b3404957c351925c46105e Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 20 Sep 2023 09:17:46 +0200
Subject: [PATCH 06/10] drop isSameType

---
 packages/php-datatypes/src/Definition/Common.php             | 5 -----
 .../php-datatypes/src/Definition/DefinitionInterface.php     | 2 --
 .../tests/unit/Backend/SourceDestinationColumnMapTest.php    | 5 -----
 3 files changed, 12 deletions(-)

diff --git a/packages/php-datatypes/src/Definition/Common.php b/packages/php-datatypes/src/Definition/Common.php
index 4bab0f25d..9e163e17a 100644
--- a/packages/php-datatypes/src/Definition/Common.php
+++ b/packages/php-datatypes/src/Definition/Common.php
@@ -168,9 +168,4 @@ protected function validateMaxLength($length, int $max, int $min = 1): bool
         }
         return (int) $length >= $min && (int) $length <= $max;
     }
-
-    public function isSameType(DefinitionInterface $definition): bool
-    {
-        return $this->type === $definition->getType();
-    }
 }
diff --git a/packages/php-datatypes/src/Definition/DefinitionInterface.php b/packages/php-datatypes/src/Definition/DefinitionInterface.php
index 385bbad8e..23d523392 100644
--- a/packages/php-datatypes/src/Definition/DefinitionInterface.php
+++ b/packages/php-datatypes/src/Definition/DefinitionInterface.php
@@ -24,6 +24,4 @@ public function isNullable(): bool;
     public function getDefault(): ?string;
 
     public static function getTypeByBasetype(string $basetype): string;
-
-    public function isSameType(DefinitionInterface $definition): bool;
 }
diff --git a/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
index 41b9d0dfc..34ca0a9b3 100644
--- a/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
+++ b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
@@ -102,11 +102,6 @@ public static function getTypeByBasetype(string $basetype): string
                     {
                         throw new Exception('Not implemented');
                     }
-
-                    public function isSameType(DefinitionInterface $definition): bool
-                    {
-                        throw new Exception('Not implemented');
-                    }
                 };
             }
 

From 0cb3cb53d46da67e40f3ab7d3819dba61dd1ee3e Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 20 Sep 2023 10:01:26 +0200
Subject: [PATCH 07/10] fix use-case

---
 .../tests/functional/Snowflake/ToFinal/SqlBuilderTest.php    | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index 22e138c65..6ea57fca6 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -841,7 +841,10 @@ public function testGetInsertAllIntoTargetTableCommandConvertToNullWithTimestamp
             convertEmptyValuesToNull: ['col1'],
             isIncremental: false,
             useTimestamp: true,
-            ignoreColumns: ['id'],
+            ignoreColumns: [
+                'id',
+                ToStageImporterInterface::TIMESTAMP_COLUMN_NAME
+            ],
         );
         $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
             $fakeStage,

From 3a61db2c8a4c5316b10d4844cf80bd8a1eef294b Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Wed, 20 Sep 2023 12:03:57 +0200
Subject: [PATCH 08/10] fix ignore multiple columns

---
 .../Backend/SourceDestinationColumnMap.php    | 44 ++++++++++++++-----
 .../Snowflake/ToFinal/SqlBuilderTest.php      |  2 +-
 .../SourceDestinationColumnMapTest.php        | 22 ++++++++++
 3 files changed, 57 insertions(+), 11 deletions(-)

diff --git a/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
index 798e2a167..b566802f0 100644
--- a/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
+++ b/packages/php-db-import-export/src/Backend/SourceDestinationColumnMap.php
@@ -6,6 +6,7 @@
 
 use Error;
 use Exception;
+use Generator;
 use Keboola\Db\ImportExport\Exception\ColumnsMismatchException;
 use Keboola\TableBackendUtils\Column\ColumnCollection;
 use Keboola\TableBackendUtils\Column\ColumnInterface;
@@ -54,18 +55,15 @@ private function buildMap(): void
         $it0 = $this->source->getIterator();
         $it1 = $this->destination->getIterator();
         while ($it0->valid() || $it1->valid()) {
-            if ($it0->valid() && in_array($it0->current()->getColumnName(), $this->ignoreColumns, true)) {
-                $it0->next();
-                if (!$it0->valid() && !$it1->valid()) {
-                    break;
-                }
+            $it0 = $this->ignoreColumn($it0, $it1);
+            if ($it0 === false) {
+                break;
             }
-            if ($it1->valid() && in_array($it1->current()->getColumnName(), $this->ignoreColumns, true)) {
-                $it1->next();
-                if (!$it0->valid() && !$it1->valid()) {
-                    break;
-                }
+            $it1 = $this->ignoreColumn($it1, $it0);
+            if ($it1 === false) {
+                break;
             }
+
             if ($it0->valid() && $it1->valid()) {
                 /** @var ColumnInterface $sourceCol */
                 $sourceCol = $it0->current();
@@ -91,4 +89,30 @@ public function getDestination(ColumnInterface $source): ColumnInterface
         assert($destination !== null);
         return $destination;
     }
+
+    /**
+     * @param Generator<int, ColumnInterface> $it0
+     * @param Generator<int, ColumnInterface> $it1
+     * @return Generator<int, ColumnInterface>|false
+     */
+    private function ignoreColumn(Generator $it0, Generator $it1): Generator|false
+    {
+        if ($this->isIgnoredColumn($it0)) {
+            $it0->next();
+            $this->ignoreColumn($it0, $it1);
+            if (!$it0->valid() && !$it1->valid()) {
+                return false;
+            }
+        }
+
+        return $it0;
+    }
+
+    /**
+     * @param Generator<int, ColumnInterface> $it
+     */
+    private function isIgnoredColumn(Generator $it): bool
+    {
+        return $it->valid() && in_array($it->current()->getColumnName(), $this->ignoreColumns, true);
+    }
 }
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index 6ea57fca6..584876818 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -843,7 +843,7 @@ public function testGetInsertAllIntoTargetTableCommandConvertToNullWithTimestamp
             useTimestamp: true,
             ignoreColumns: [
                 'id',
-                ToStageImporterInterface::TIMESTAMP_COLUMN_NAME
+                ToStageImporterInterface::TIMESTAMP_COLUMN_NAME,
             ],
         );
         $sql = $this->getBuilder()->getInsertAllIntoTargetTableCommand(
diff --git a/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
index 34ca0a9b3..353b924db 100644
--- a/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
+++ b/packages/php-db-import-export/tests/unit/Backend/SourceDestinationColumnMapTest.php
@@ -195,6 +195,28 @@ public function testIgnoreColumn(): void
         );
     }
 
+    public function testIgnoreColumnsMoreThanOne(): void
+    {
+        $this->expectNotToPerformAssertions();
+        $source = new ColumnCollection([
+            $this->getColumn('col1', 'string'),
+            $this->getColumn('col2', 'string'),
+        ]);
+        $destination = new ColumnCollection([
+            $this->getColumn('col1', 'bool'),
+            $this->getColumn('col2', 'bool'),
+            $this->getColumn('col3', 'bool'),
+            $this->getColumn('col4', 'bool'),
+            $this->getColumn('col5', 'bool'),
+        ]);
+
+        new SourceDestinationColumnMap(
+            $source,
+            $destination,
+            ['col3', 'col4', 'col5']
+        );
+    }
+
     public function testColumnNotFound(): void
     {
         [, , $map] = $this->getMap();

From 46b5ffda96893d9b9d54e00d73c895eab64c4191 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Fri, 22 Sep 2023 10:47:03 +0200
Subject: [PATCH 09/10] remove autoCastTypes option, update test add GEOGRAPHY,
 GEOMETRY type

---
 .../Snowflake/SnowflakeImportOptions.php      | 16 ---------
 .../Snowflake/ToFinalTable/SqlBuilder.php     |  9 +++--
 .../Snowflake/ToFinal/FullImportTest.php      | 23 +++++--------
 .../ToFinal/IncrementalImportTest.php         | 33 +++++++++----------
 .../Snowflake/ToFinal/SqlBuilderTest.php      | 18 ++--------
 5 files changed, 34 insertions(+), 65 deletions(-)

diff --git a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
index 896e432e6..58ecc7bb8 100644
--- a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
+++ b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeImportOptions.php
@@ -8,11 +8,6 @@
 
 class SnowflakeImportOptions extends ImportOptions
 {
-    /**
-     * @var string[]
-     */
-    private array $autoCastTypes;
-
     /** @var self::SAME_TABLES_* */
     private bool $requireSameTables;
 
@@ -24,7 +19,6 @@ class SnowflakeImportOptions extends ImportOptions
      * @param self::SAME_TABLES_* $requireSameTables
      * @param self::NULL_MANIPULATION_* $nullManipulation
      * @param string[] $ignoreColumns
-     * @param string[] $autoCastTypes
      */
     public function __construct(
         array $convertEmptyValuesToNull = [],
@@ -34,7 +28,6 @@ public function __construct(
         bool $requireSameTables = self::SAME_TABLES_NOT_REQUIRED,
         bool $nullManipulation = self::NULL_MANIPULATION_ENABLED,
         array $ignoreColumns = [],
-        array $autoCastTypes = [],
     ) {
         parent::__construct(
             $convertEmptyValuesToNull,
@@ -46,7 +39,6 @@ public function __construct(
         );
         $this->requireSameTables = $requireSameTables;
         $this->nullManipulation = $nullManipulation;
-        $this->autoCastTypes = $autoCastTypes;
     }
 
     public function isRequireSameTables(): bool
@@ -58,12 +50,4 @@ public function isNullManipulationEnabled(): bool
     {
         return $this->nullManipulation === self::NULL_MANIPULATION_ENABLED;
     }
-
-    /**
-     * @return string[]
-     */
-    public function autoCastTypes(): array
-    {
-        return $this->autoCastTypes;
-    }
 }
diff --git a/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php b/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
index 24f45108b..4b373f992 100644
--- a/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
+++ b/packages/php-db-import-export/src/Backend/Snowflake/ToFinalTable/SqlBuilder.php
@@ -16,6 +16,11 @@
 
 class SqlBuilder
 {
+    private const AUTO_CASTING_TYPES = [
+        Snowflake::TYPE_VARIANT,
+        Snowflake::TYPE_OBJECT,
+        Snowflake::TYPE_ARRAY,
+    ];
     public const SRC_ALIAS = 'src';
 
     public function getBeginTransaction(): string
@@ -192,7 +197,7 @@ public function getInsertAllIntoTargetTableCommand(
             if (!$importOptions->isNullManipulationEnabled()) {
                 $destinationColumn = $columnMap->getDestination($sourceColumn);
                 $type = $destinationColumn->getColumnDefinition()->getType();
-                $useAutoCast = in_array($type, $importOptions->autoCastTypes(), true);
+                $useAutoCast = in_array($type, self::AUTO_CASTING_TYPES, true);
                 $isSameType = $type === $sourceColumn->getColumnDefinition()->getType();
                 if ($useAutoCast && !$isSameType) {
                     if ($type === Snowflake::TYPE_OBJECT) {
@@ -292,7 +297,7 @@ public function getUpdateWithPkCommand(
             if (!$importOptions->isNullManipulationEnabled()) {
                 $destinationColumn = $columnMap->getDestination($sourceColumn);
                 $type = $destinationColumn->getColumnDefinition()->getType();
-                $useAutoCast = in_array($type, $importOptions->autoCastTypes(), true);
+                $useAutoCast = in_array($type, self::AUTO_CASTING_TYPES, true);
                 $isSameType = $type === $sourceColumn->getColumnDefinition()->getType();
                 if ($useAutoCast && !$isSameType) {
                     if ($type === Snowflake::TYPE_OBJECT) {
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
index be2443455..3bebf9dfb 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/FullImportTest.php
@@ -39,10 +39,6 @@ protected function setUp(): void
     /**
      * Test is testing loading of semi-structured data into typed table.
      *
-     * We ignore here GEOGRAPHY and GEOMETRY as they act differently when casting from string
-     * https://docs.snowflake.com/en/sql-reference/functions/to_geography
-     * https://docs.snowflake.com/en/sql-reference/functions/to_geometry
-     *
      * This test is not using CSV but inserting data directly into stage table to mimic this behavior
      */
     public function testLoadTypedTableWithCastingValues(): void
@@ -56,6 +52,8 @@ public function testLoadTypedTableWithCastingValues(): void
               "VARBINARY" VARBINARY,
               "OBJECT" OBJECT,
               "ARRAY" ARRAY,
+              "GEOGRAPHY" GEOGRAPHY,
+              "GEOMETRY" GEOMETRY,
               "_timestamp" TIMESTAMP
             );',
             SnowflakeQuote::quoteSingleIdentifier($this->getDestinationSchemaName())
@@ -69,14 +67,7 @@ public function testLoadTypedTableWithCastingValues(): void
             1,
             SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
             SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
-            [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
-            [
-                Snowflake::TYPE_VARIANT,
-                Snowflake::TYPE_BINARY,
-                Snowflake::TYPE_VARBINARY,
-                Snowflake::TYPE_OBJECT,
-                Snowflake::TYPE_ARRAY,
-            ]
+            [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME]
         );
 
         $destinationRef = new SnowflakeTableReflection(
@@ -95,6 +86,8 @@ public function testLoadTypedTableWithCastingValues(): void
                 'VARBINARY',
                 'OBJECT',
                 'ARRAY',
+                'GEOGRAPHY',
+                'GEOMETRY',
             ]
         );
 
@@ -104,13 +97,15 @@ public function testLoadTypedTableWithCastingValues(): void
         );
         $this->connection->executeQuery(sprintf(
         /** @lang Snowflake */
-            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY","GEOGRAPHY","GEOMETRY") 
 select 1, 
        TO_VARCHAR(TO_VARIANT(\'3.14\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
-       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL)),
+       \'POINT(-122.35 37.55)\',
+       \'POINT(1820.12 890.56)\'
 ;',
             $stagingTable->getSchemaName(),
             $stagingTable->getTableName()
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
index 1d7002d3b..0e139d042 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/IncrementalImportTest.php
@@ -50,10 +50,6 @@ protected function setUp(): void
     /**
      * Test is testing loading of semi-structured data into typed table.
      *
-     * We ignore here GEOGRAPHY and GEOMETRY as they act differently when casting from string
-     * https://docs.snowflake.com/en/sql-reference/functions/to_geography
-     * https://docs.snowflake.com/en/sql-reference/functions/to_geometry
-     *
      * This test is not using CSV but inserting data directly into stage table to mimic this behavior
      */
     public function testLoadTypedTableWithCastingValues(): void
@@ -67,6 +63,8 @@ public function testLoadTypedTableWithCastingValues(): void
               "VARBINARY" VARBINARY,
               "OBJECT" OBJECT,
               "ARRAY" ARRAY,
+              "GEOGRAPHY" GEOGRAPHY,
+              "GEOMETRY" GEOMETRY,
               "_timestamp" TIMESTAMP,
                PRIMARY KEY ("id")
             );',
@@ -74,13 +72,15 @@ public function testLoadTypedTableWithCastingValues(): void
         ));
         $this->connection->executeQuery(sprintf(
         /** @lang Snowflake */
-            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY","GEOGRAPHY","GEOMETRY") 
 SELECT 1, 
       TO_VARIANT(\'3.14\'),
       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
       TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\'),
       OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT),
-      ARRAY_CONSTRUCT(1, 2, 3, NULL)
+      ARRAY_CONSTRUCT(1, 2, 3, NULL),
+      \'POINT(-122.35 37.55)\',
+      \'POINT(1820.12 890.56)\'
 ;',
             $this->getDestinationSchemaName(),
             'types'
@@ -95,13 +95,6 @@ public function testLoadTypedTableWithCastingValues(): void
             requireSameTables: SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
             nullManipulation: SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
             ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
-            autoCastTypes: [
-                Snowflake::TYPE_VARIANT,
-                Snowflake::TYPE_BINARY,
-                Snowflake::TYPE_VARBINARY,
-                Snowflake::TYPE_OBJECT,
-                Snowflake::TYPE_ARRAY,
-            ]
         );
 
         $destinationRef = new SnowflakeTableReflection(
@@ -120,6 +113,8 @@ public function testLoadTypedTableWithCastingValues(): void
                 'VARBINARY',
                 'OBJECT',
                 'ARRAY',
+                'GEOGRAPHY',
+                'GEOMETRY',
             ]
         );
 
@@ -129,26 +124,30 @@ public function testLoadTypedTableWithCastingValues(): void
         );
         $this->connection->executeQuery(sprintf(
         /** @lang Snowflake */
-            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY","GEOGRAPHY","GEOMETRY") 
 SELECT 1, 
        TO_VARCHAR(TO_VARIANT(\'3.14\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
-       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL)),
+       \'POINT(-122.35 37.55)\',
+       \'POINT(1820.12 890.56)\'
 ;',
             $stagingTable->getSchemaName(),
             $stagingTable->getTableName()
         ));
         $this->connection->executeQuery(sprintf(
         /** @lang Snowflake */
-            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY") 
+            'INSERT INTO "%s"."%s" ("id","VARIANT","BINARY","VARBINARY","OBJECT","ARRAY","GEOGRAPHY","GEOMETRY") 
 SELECT 2, 
        TO_VARCHAR(TO_VARIANT(\'3.14\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(TO_BINARY(HEX_ENCODE(\'1\'), \'HEX\')),
        TO_VARCHAR(OBJECT_CONSTRUCT(\'name\', \'Jones\'::VARIANT, \'age\',  42::VARIANT)),
-       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL))
+       TO_VARCHAR(ARRAY_CONSTRUCT(1, 2, 3, NULL)),
+       \'POINT(-122.35 37.55)\',
+       \'POINT(1820.12 890.56)\'
 ;',
             $stagingTable->getSchemaName(),
             $stagingTable->getTableName()
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
index 584876818..dbc468193 100644
--- a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/SqlBuilderTest.php
@@ -559,20 +559,13 @@ public function testGetInsertAllIntoTargetTableCommandCasting(): void
                 requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
                 nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
                 ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
-                autoCastTypes: [
-                    Snowflake::TYPE_VARIANT,
-                    Snowflake::TYPE_BINARY,
-                    Snowflake::TYPE_VARBINARY,
-                    Snowflake::TYPE_OBJECT,
-                    Snowflake::TYPE_ARRAY,
-                ],
             ),
             '2020-01-01 00:00:00'
         );
 
         self::assertEquals(
         // phpcs:ignore
-            'INSERT INTO "import_export_test_schema"."import_export_test_test" ("pk1", "VARIANT", "BINARY", "VARBINARY", "OBJECT", "ARRAY") (SELECT "pk1",CAST("VARIANT" AS VARIANT) AS "VARIANT",CAST("BINARY" AS BINARY) AS "BINARY",CAST("VARBINARY" AS VARBINARY) AS "VARBINARY",CAST(TO_VARIANT("OBJECT") AS OBJECT) AS "OBJECT",CAST("ARRAY" AS ARRAY) AS "ARRAY" FROM "import_export_test_schema"."__temp_stagingTable" AS "src")',
+            'INSERT INTO "import_export_test_schema"."import_export_test_test" ("pk1", "VARIANT", "BINARY", "VARBINARY", "OBJECT", "ARRAY") (SELECT "pk1",CAST("VARIANT" AS VARIANT) AS "VARIANT","BINARY","VARBINARY",CAST(TO_VARIANT("OBJECT") AS OBJECT) AS "OBJECT",CAST("ARRAY" AS ARRAY) AS "ARRAY" FROM "import_export_test_schema"."__temp_stagingTable" AS "src")',
             $sql
         );
 
@@ -1153,19 +1146,12 @@ public function testGetUpdateWithPkCommandCasting(): void
                 requireSameTables: ImportOptions::SAME_TABLES_NOT_REQUIRED,
                 nullManipulation: ImportOptions::NULL_MANIPULATION_SKIP, //<- skipp null manipulation
                 ignoreColumns: [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME],
-                autoCastTypes: [
-                    Snowflake::TYPE_VARIANT,
-                    Snowflake::TYPE_BINARY,
-                    Snowflake::TYPE_VARBINARY,
-                    Snowflake::TYPE_OBJECT,
-                    Snowflake::TYPE_ARRAY,
-                ],
             ),
             '2020-01-01 00:00:00'
         );
         self::assertEquals(
         // phpcs:ignore
-            'UPDATE "import_export_test_schema"."import_export_test_test" AS "dest" SET "pk1" = "src"."pk1", "VARIANT" = CAST("src"."VARIANT" AS VARIANT), "BINARY" = CAST("src"."BINARY" AS BINARY), "VARBINARY" = CAST("src"."VARBINARY" AS VARBINARY), "OBJECT" = CAST(TO_VARIANT("src"."OBJECT") AS OBJECT), "ARRAY" = CAST("src"."ARRAY" AS ARRAY) FROM "import_export_test_schema"."__temp_stagingTable" AS "src" WHERE "dest"."pk1" = "src"."pk1" ',
+            'UPDATE "import_export_test_schema"."import_export_test_test" AS "dest" SET "pk1" = "src"."pk1", "VARIANT" = CAST("src"."VARIANT" AS VARIANT), "BINARY" = "src"."BINARY", "VARBINARY" = "src"."VARBINARY", "OBJECT" = CAST(TO_VARIANT("src"."OBJECT") AS OBJECT), "ARRAY" = CAST("src"."ARRAY" AS ARRAY) FROM "import_export_test_schema"."__temp_stagingTable" AS "src" WHERE "dest"."pk1" = "src"."pk1" ',
             $sql
         );
         $this->connection->executeStatement($sql);

From d5347180eeda26a06b18c27a1805583018be2d21 Mon Sep 17 00:00:00 2001
From: zajca <zajca@zajca.cz>
Date: Fri, 22 Sep 2023 11:46:35 +0200
Subject: [PATCH 10/10] add tests for casting exceptions

---
 .../Backend/Snowflake/SnowflakeException.php  |  21 ++-
 .../Snowflake/SnowflakeExceptionTest.php      |  30 +++-
 .../ToFinal/StageToFinalCastingErrorsTest.php | 137 ++++++++++++++++++
 3 files changed, 182 insertions(+), 6 deletions(-)
 create mode 100644 packages/php-db-import-export/tests/functional/Snowflake/ToFinal/StageToFinalCastingErrorsTest.php

diff --git a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeException.php b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeException.php
index 5745cdaed..c07dd15ab 100644
--- a/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeException.php
+++ b/packages/php-db-import-export/src/Backend/Snowflake/SnowflakeException.php
@@ -31,11 +31,24 @@ public static function covertException(Throwable $e): Throwable
         }
 
         // phpcs:ignore
-        $isNullInNotNullCol = preg_match('/NULL result in a non-nullable column/', $e->getMessage(), $output_array) === 1;
-        $isNotRecognized = preg_match('/ \'(.*)\' is not recognized/', $e->getMessage(), $output_array) === 1;
-        if ($isNullInNotNullCol || $isNotRecognized) {
+        $message = $e->getMessage();
+        $isObjectCastFail = preg_match('/Failed to cast variant value .* to OBJECT/', $message, $output_array) === 1;
+        if ($isObjectCastFail) {
+            // remove variant from message as it would confuse users
+            // we are using TO_OBJECT(TO_VARIANT(...)) casting combination
+            $message = str_replace('variant ', '', $message);
+        }
+        $isInvalidGeo = preg_match('/Error parsing Geo input/', $message, $output_array) === 1;
+        // phpcs:ignore
+        $isInvalidBinary = preg_match('/The following string is not a legal hex-encoded value/', $message, $output_array) === 1;
+        $isNullInNotNullCol = preg_match('/NULL result in a non-nullable column/', $message, $output_array) === 1;
+        $isNotRecognized = preg_match('/ \'(.*)\' is not recognized/', $message, $output_array) === 1;
+        if ($isNotRecognized) {
+            $message .= '. Value you are trying to load cannot be converted to used datatype.';
+        }
+        if ($isNullInNotNullCol || $isNotRecognized || $isInvalidBinary || $isInvalidGeo || $isObjectCastFail) {
             return new Exception(
-                'Load error: ' . $e->getMessage(),
+                'Load error: ' . $message,
                 Exception::VALUE_CONVERSION,
                 $e
             );
diff --git a/packages/php-db-import-export/tests/functional/Backend/Snowflake/SnowflakeExceptionTest.php b/packages/php-db-import-export/tests/functional/Backend/Snowflake/SnowflakeExceptionTest.php
index b3a7e0f02..6157a661c 100644
--- a/packages/php-db-import-export/tests/functional/Backend/Snowflake/SnowflakeExceptionTest.php
+++ b/packages/php-db-import-export/tests/functional/Backend/Snowflake/SnowflakeExceptionTest.php
@@ -58,7 +58,8 @@ public function provideExceptions(): Generator
         yield 'value conversion' => [
             "Numeric value 'male' is not recognized",
             ImportException::class,
-            "Load error: Numeric value 'male' is not recognized",
+            // phpcs:ignore
+            "Load error: Numeric value 'male' is not recognized. Value you are trying to load cannot be converted to used datatype.",
             13, // VALUE_CONVERSION
             true,
         ];
@@ -66,7 +67,8 @@ public function provideExceptions(): Generator
         yield 'value conversion 2' => [
             "Numeric value 'ma\'le' is not recognized",
             ImportException::class,
-            "Load error: Numeric value 'ma\'le' is not recognized",
+            // phpcs:ignore
+            "Load error: Numeric value 'ma\'le' is not recognized. Value you are trying to load cannot be converted to used datatype.",
             13, // VALUE_CONVERSION
             true,
         ];
@@ -108,5 +110,29 @@ public function provideExceptions(): Generator
             11, // ROW_SIZE_TOO_LARGE
             true,
         ];
+
+        yield 'GEO casting error' => [
+            'Error parsing Geo input: xxx. Did not recognize valid GeoJSON, (E)WKT or (E)WKB.',
+            ImportException::class,
+            'Load error: Error parsing Geo input: xxx. Did not recognize valid GeoJSON, (E)WKT or (E)WKB.',
+            13, // VALUE_CONVERSION
+            true,
+        ];
+
+        yield 'OBJECT casting error' => [
+            'An exception occurred while executing a query: Failed to cast variant value "xxx" to OBJECT',
+            ImportException::class,
+            'Load error: An exception occurred while executing a query: Failed to cast value "xxx" to OBJECT',
+            13, // VALUE_CONVERSION
+            true,
+        ];
+
+        yield 'Binary casting error' => [
+            "The following string is not a legal hex-encoded value: 'xxx'",
+            ImportException::class,
+            "Load error: The following string is not a legal hex-encoded value: 'xxx'",
+            13, // VALUE_CONVERSION
+            true,
+        ];
     }
 }
diff --git a/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/StageToFinalCastingErrorsTest.php b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/StageToFinalCastingErrorsTest.php
new file mode 100644
index 000000000..ec1e5f7da
--- /dev/null
+++ b/packages/php-db-import-export/tests/functional/Snowflake/ToFinal/StageToFinalCastingErrorsTest.php
@@ -0,0 +1,137 @@
+<?php
+
+declare(strict_types=1);
+
+namespace Tests\Keboola\Db\ImportExportFunctional\Snowflake\ToFinal;
+
+use Generator;
+use Keboola\Datatype\Definition\Snowflake;
+use Keboola\Db\Import\Exception;
+use Keboola\Db\ImportExport\Backend\ImportState;
+use Keboola\Db\ImportExport\Backend\Snowflake\SnowflakeImportOptions;
+use Keboola\Db\ImportExport\Backend\Snowflake\ToFinalTable\FullImporter;
+use Keboola\Db\ImportExport\Backend\Snowflake\ToStage\StageTableDefinitionFactory;
+use Keboola\Db\ImportExport\Backend\ToStageImporterInterface;
+use Keboola\TableBackendUtils\Column\Snowflake\SnowflakeColumn;
+use Keboola\TableBackendUtils\Escaping\Snowflake\SnowflakeQuote;
+use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableDefinition;
+use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableQueryBuilder;
+use Keboola\TableBackendUtils\Table\Snowflake\SnowflakeTableReflection;
+use Tests\Keboola\Db\ImportExportFunctional\Snowflake\SnowflakeBaseTestCase;
+
+class StageToFinalCastingErrorsTest extends SnowflakeBaseTestCase
+{
+    protected function setUp(): void
+    {
+        parent::setUp();
+        $this->cleanSchema($this->getDestinationSchemaName());
+        $this->cleanSchema($this->getSourceSchemaName());
+        $this->createSchema($this->getSourceSchemaName());
+        $this->createSchema($this->getDestinationSchemaName());
+    }
+
+    public function castingErrorCases(): Generator
+    {
+        yield 'BINARY string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_BINARY)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/The following string is not a legal hex-encoded value/',
+        ];
+        yield 'VARBINARY string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_VARBINARY)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/The following string is not a legal hex-encoded value/',
+        ];
+        yield 'OBJECT string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_OBJECT)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/Failed to cast variant value .* to OBJECT/',
+        ];
+        yield 'GEOGRAPHY string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_GEOGRAPHY)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/Error parsing Geo input/',
+        ];
+        yield 'GEOMETRY string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_GEOMETRY)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/Error parsing Geo input/',
+        ];
+        yield 'TIMESTAMP string' => [
+            'column' => new SnowflakeColumn('id', new Snowflake(Snowflake::TYPE_TIMESTAMP)),
+            'insertData' => '\'xxx\'',
+            'expectedMessage' => '/is not recognized/',
+        ];
+    }
+
+    /**
+     * @dataProvider castingErrorCases
+     */
+    public function testLoadTypedTableWithCastingValuesErrors(
+        SnowflakeColumn $column,
+        string $insertData,
+        string $expectedMessage,
+    ): void {
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'CREATE TABLE %s."types" (
+              "%s" %s,
+              "_timestamp" TIMESTAMP
+            );',
+            SnowflakeQuote::quoteSingleIdentifier($this->getDestinationSchemaName()),
+            $column->getColumnName(),
+            $column->getColumnDefinition()->getSQLDefinition(),
+        ));
+
+        // skipping header
+        $options = new SnowflakeImportOptions(
+            [],
+            false,
+            false,
+            1,
+            SnowflakeImportOptions::SAME_TABLES_NOT_REQUIRED,
+            SnowflakeImportOptions::NULL_MANIPULATION_SKIP,
+            [ToStageImporterInterface::TIMESTAMP_COLUMN_NAME]
+        );
+
+        $destinationRef = new SnowflakeTableReflection(
+            $this->connection,
+            $this->getDestinationSchemaName(),
+            'types'
+        );
+        /** @var SnowflakeTableDefinition $destination */
+        $destination = $destinationRef->getTableDefinition();
+        $stagingTable = StageTableDefinitionFactory::createVarcharStagingTableDefinition(
+            $destination->getSchemaName(),
+            [
+                $column->getColumnName(),
+            ]
+        );
+
+        $qb = new SnowflakeTableQueryBuilder();
+        $this->connection->executeStatement(
+            $qb->getCreateTableCommandFromDefinition($stagingTable)
+        );
+        $this->connection->executeQuery(sprintf(
+        /** @lang Snowflake */
+            'INSERT INTO "%s"."%s" ("%s") SELECT %s;',
+            $stagingTable->getSchemaName(),
+            $stagingTable->getTableName(),
+            $column->getColumnName(),
+            $insertData
+        ));
+        $toFinalTableImporter = new FullImporter($this->connection);
+
+        try {
+            $toFinalTableImporter->importToTable(
+                $stagingTable,
+                $destination,
+                $options,
+                new ImportState($stagingTable->getTableName())
+            );
+            $this->fail('Import must fail');
+        } catch (Exception $e) {
+            $this->assertMatchesRegularExpression($expectedMessage, $e->getMessage());
+        }
+    }
+}