//nolint:dupl,nolintlint //TODO: fix this

package sql_system //nolint:revive,stylecheck // package name is meaningful and readable

import (

"database/sql"

"fmt"

"reflect"

"regexp"

"strings"

"time"

"github.com/lib/pq/oid"

"github.com/stackql/any-sdk/pkg/constants"

"github.com/stackql/any-sdk/pkg/db/sqlcontrol"

"github.com/stackql/any-sdk/pkg/dto"

"github.com/stackql/any-sdk/pkg/logging"

"github.com/stackql/any-sdk/public/formulation"

"github.com/stackql/any-sdk/public/sqlengine"

"github.com/stackql/stackql-parser/go/vt/sqlparser"

"github.com/stackql/stackql/internal/stackql/astfuncrewrite"

"github.com/stackql/stackql/internal/stackql/internal_data_transfer/internaldto"

"github.com/stackql/stackql/internal/stackql/internal_data_transfer/relationaldto"

"github.com/stackql/stackql/internal/stackql/typing"

"github.com/stackql/stackql/pkg/serde"

"github.com/stackql/stackql/pkg/textutil"

)

func newSQLiteSystem(

sqlEngine sqlengine.SQLEngine,

analyticsNamespaceLikeString string,

controlAttributes sqlcontrol.ControlAttributes,

formatter sqlparser.NodeFormatter,

sqlCfg dto.SQLBackendCfg, //nolint:unparam,revive // future proof

authCfg map[string]*dto.AuthCtx,

typCfg typing.Config,

exportNamepsace string,

) (SQLSystem, error) {

rv := &sqLiteSystem{

defaultGolangKind: reflect.String,

defaultRelationalType: "text",

typeCfg: typCfg,

controlAttributes: controlAttributes,

analyticsNamespaceLikeString: analyticsNamespaceLikeString,

sqlEngine: sqlEngine,

formatter: formatter,

authCfg: authCfg,

exportNamespace: exportNamepsace,

}

err := rv.initSQLiteEngine()

return rv, err

}

type sqLiteSystem struct {

controlAttributes sqlcontrol.ControlAttributes

analyticsNamespaceLikeString string

sqlEngine sqlengine.SQLEngine

formatter sqlparser.NodeFormatter

typeCfg typing.Config

defaultRelationalType string

defaultGolangKind reflect.Kind

authCfg map[string]*dto.AuthCtx

exportNamespace string

}

func (eng *sqLiteSystem) initSQLiteEngine() error {

_, err := eng.sqlEngine.Exec(sqLiteEngineSetupDDL)

return err

}

func (eng *sqLiteSystem) GetTable(

tableHeirarchyIDs internaldto.HeirarchyIdentifiers,

discoveryID int,

) (internaldto.DBTable, error) {

return eng.getTable(tableHeirarchyIDs, discoveryID)

}

func (eng *sqLiteSystem) getTable(

tableHeirarchyIDs internaldto.HeirarchyIdentifiers,

discoveryID int,

) (internaldto.DBTable, error) {

tableNameStump, err := eng.getTableNameStump(tableHeirarchyIDs)

if err != nil {

return internaldto.NewDBTable("", "", "", 0, tableHeirarchyIDs), err

}

tableName := fmt.Sprintf("%s.generation_%d", tableNameStump, discoveryID)

return internaldto.NewDBTable(

tableName,

tableNameStump,

tableHeirarchyIDs.GetTableName(),

discoveryID,

tableHeirarchyIDs), err

}

func (eng *sqLiteSystem) GetCurrentTable(

tableHeirarchyIDs internaldto.HeirarchyIdentifiers,

) (internaldto.DBTable, error) {

return eng.getCurrentTable(tableHeirarchyIDs)

}

//nolint:unparam // future proof

func (eng *sqLiteSystem) getTableNameStump(tableHeirarchyIDs internaldto.HeirarchyIdentifiers) (string, error) {

return tableHeirarchyIDs.GetTableName(), nil

}

func (eng *sqLiteSystem) getCurrentTable(

tableHeirarchyIDs internaldto.HeirarchyIdentifiers,

) (internaldto.DBTable, error) {

var tableName string

var discoID int

tableNameStump, err := eng.getTableNameStump(tableHeirarchyIDs)

if err != nil {

return internaldto.NewDBTable("", "", "", 0, tableHeirarchyIDs), err

}

if _, isView := tableHeirarchyIDs.GetView(); isView {

return internaldto.NewDBTable(

tableNameStump,

tableNameStump,

tableHeirarchyIDs.GetTableName(),

discoID,

tableHeirarchyIDs,

), nil

}

tableNamePattern := fmt.Sprintf("%s.generation_%%", tableNameStump)

tableNameLHSRemove := fmt.Sprintf("%s.generation_", tableNameStump)

res := eng.sqlEngine.QueryRow(

`select name, CAST(REPLACE(name, ?, '') AS INTEGER) from sqlite_schema where type = 'table' and name like ? ORDER BY name DESC limit 1`, // nolint:lll // this is a long query, but it's a single line

tableNameLHSRemove,

tableNamePattern,

)

err = res.Scan(&tableName, &discoID)

if err != nil {

logging.GetLogger().Errorln(

fmt.Sprintf("err = %v for tableNamePattern = '%s' and tableNameLHSRemove = '%s'",

err,

tableNamePattern,

tableNameLHSRemove,

),

)

}

return internaldto.NewDBTable(

tableName,

tableNameStump,

tableHeirarchyIDs.GetTableName(),

discoID,

tableHeirarchyIDs,

), nil

}

func (eng *sqLiteSystem) GetName() string {

return constants.SQLDialectSQLite3

}

func (eng *sqLiteSystem) GetASTFormatter() sqlparser.NodeFormatter {

return eng.formatter

}

func (eng *sqLiteSystem) GetASTFuncRewriter() astfuncrewrite.ASTFuncRewriter {

return astfuncrewrite.GetNopFuncRewriter()

}

//nolint:revive // future proof

func (eng *sqLiteSystem) GCAdd(

tableName string, parentTcc,

lockableTcc internaldto.TxnControlCounters,

) error {

maxTxnColName := eng.controlAttributes.GetControlMaxTxnColumnName()

q := fmt.Sprintf(

`

UPDATE "%s"

SET "%s" = r.current_value

FROM (

SELECT *

FROM

"__iql__.control.gc.rings"

) AS r

WHERE

"%s" = ?

AND

"%s" = ?

AND

r.ring_name = 'transaction_id'

AND

"%s" < CASE

WHEN ("%s" - r.current_offset) < 0

THEN CAST(pow(2, r.width_bits) + ("%s" - r.current_offset) AS int)

ELSE "%s" - r.current_offset

END

`,

tableName,

maxTxnColName,

eng.controlAttributes.GetControlTxnIDColumnName(),

eng.controlAttributes.GetControlInsIDColumnName(),

maxTxnColName,

maxTxnColName,

maxTxnColName,

maxTxnColName,

)

_, err := eng.sqlEngine.Exec(q, lockableTcc.GetTxnID(), lockableTcc.GetInsertID())

return err

}

func (eng *sqLiteSystem) GCCollectObsoleted(minTransactionID int) error {

return eng.gCCollectObsoleted(minTransactionID)

}

func (eng *sqLiteSystem) RegisterExternalTable(

connectionName string,

tableDetails formulation.SQLExternalTable,

) error {

return eng.registerExternalTable(connectionName, tableDetails)

}

func (eng *sqLiteSystem) registerExternalTable(

connectionName string,

tableDetails formulation.SQLExternalTable,

) error {

q := `

INSERT OR IGNORE INTO "__iql__.external.columns" (

connection_name

,catalog_name

,schema_name

,table_name

,column_name

,column_type

,ordinal_position

,"oid"

,column_width

,column_precision

) VALUES (

?

,?

,?

,?

,?

,?

,?

,?

,?

,?

)

`

tx, err := eng.sqlEngine.GetTx()

if err != nil {

return err

}

for ord, col := range tableDetails.GetColumns() {

_, err = tx.Exec(

q,

connectionName,

tableDetails.GetCatalogName(),

tableDetails.GetSchemaName(),

tableDetails.GetName(),

col.GetName(),

col.GetType(),

ord,

col.GetOid(),

col.GetWidth(),

col.GetPrecision(),

)

if err != nil {

//nolint:errcheck // TODO: merge variadic error(s) into one

tx.Rollback()

return err

}

}

err = tx.Commit()

return err

}

func (eng *sqLiteSystem) ObtainRelationalColumnsFromExternalSQLtable(

hierarchyIDs internaldto.HeirarchyIdentifiers,

) ([]typing.RelationalColumn, error) {

return eng.obtainRelationalColumnsFromExternalSQLtable(hierarchyIDs)

}

func (eng *sqLiteSystem) ObtainRelationalColumnFromExternalSQLtable(

hierarchyIDs internaldto.HeirarchyIdentifiers,

colName string,

) (typing.RelationalColumn, error) {

return eng.obtainRelationalColumnFromExternalSQLtable(hierarchyIDs, colName)

}

func (eng *sqLiteSystem) getSQLExternalSchema(providerName string) string {

rv := ""

if eng.authCfg != nil {

ac, ok := eng.authCfg[providerName]

if ok && ac != nil {

sqlCfg, sqlOk := ac.GetSQLCfg()

if sqlOk {

rv = sqlCfg.GetSchemaType()

}

}

}

if rv == "" {

rv = constants.SQLDataSourceSchemaDefault

}

return rv

}

//nolint:gosec // who cares

func (eng *sqLiteSystem) obtainRelationalColumnsFromExternalSQLtable(

hierarchyIDs internaldto.HeirarchyIdentifiers,

) ([]typing.RelationalColumn, error) {

q := `

SELECT

column_name

,column_type

,"oid"

,column_width

,column_precision

FROM

"__iql__.external.columns"

WHERE

connection_name = ?

AND

catalog_name = ?

AND

schema_name = ?

AND

table_name = ?

ORDER BY ordinal_position ASC

`

providerName := hierarchyIDs.GetProviderStr()

connectionName := eng.getSQLExternalSchema(providerName)

catalogName := ""

schemaName := hierarchyIDs.GetServiceStr()

tableName := hierarchyIDs.GetResourceStr()

rows, err := eng.sqlEngine.Query( //nolint:rowserrcheck // TODO: fix this

q,

connectionName,

catalogName,

schemaName,

tableName,

)

if err != nil {

return nil, err

}

defer rows.Close()

hasRow := false

var rv []typing.RelationalColumn

for {

if !rows.Next() {

break

}

hasRow = true

var columnName, columnType string

var oID, colWidth, colPrecision int

err = rows.Scan(&columnName, &columnType, &oID, &colWidth, &colPrecision)

if err != nil {

return nil, err

}

relationalColumn := typing.NewRelationalColumn(

columnName,

columnType).WithWidth(colWidth).WithOID(oid.Oid(oID))

rv = append(rv, relationalColumn)

}

if !hasRow {

return nil, fmt.Errorf(

"cannot generate relational table from external table = '%s': not present in external metadata",

tableName,

)

}

return rv, nil

}

//nolint:gosec // TODO: establish pattern

func (eng *sqLiteSystem) obtainRelationalColumnFromExternalSQLtable(

hierarchyIDs internaldto.HeirarchyIdentifiers,

colName string,

) (typing.RelationalColumn, error) {

q := `

SELECT

column_name

,column_type

,"oid"

,column_width

,column_precision

FROM

"__iql__.external.columns"

WHERE

connection_name = ?

AND

catalog_name = ?

AND

schema_name = ?

AND

table_name = ?

AND

column_name = ?

ORDER BY ordinal_position ASC

`

providerName := hierarchyIDs.GetProviderStr()

connectionName := eng.getSQLExternalSchema(providerName)

catalogName := ""

schemaName := hierarchyIDs.GetServiceStr()

tableName := hierarchyIDs.GetResourceStr()

row := eng.sqlEngine.QueryRow(

q,

connectionName,

catalogName,

schemaName,

tableName,

colName,

)

var columnName, columnType string

var oID, colWidth, colPrecision int

err := row.Scan(&columnName, &columnType, &oID, &colWidth, &colPrecision)

if err != nil {

return nil, err

}

relationalColumn := typing.NewRelationalColumn(columnName, columnType).WithWidth(colWidth).WithOID(oid.Oid(oID))

return relationalColumn, nil

}

// Garbage collection and purge act on provider cache tables only. The physical

// tables behind user space tables and materialized views are user data and are

// excluded via the "__iql__.tables" and "__iql__.materialized_views" catalogues

// (issue #799).

func (eng *sqLiteSystem) gCCollectObsoleted(minTransactionID int) error {

maxTxnColName := eng.controlAttributes.GetControlMaxTxnColumnName()

obtainQuery := fmt.Sprintf(

`

SELECT

'DELETE FROM "' || name || '" WHERE "%s" < %d ; '

FROM

sqlite_master

where

type = 'table'

and

name not like '__iql__%%'

and

name NOT LIKE 'sqlite_%%'

and

name NOT IN (SELECT view_name FROM "__iql__.materialized_views")

and

name NOT IN (SELECT table_name FROM "__iql__.tables")

`,

maxTxnColName,

minTransactionID,

)

deleteQueryResultSet, err := eng.sqlEngine.Query(obtainQuery)

if err != nil {

return err

}

return eng.readExecGeneratedQueries(deleteQueryResultSet)

}

func (eng *sqLiteSystem) GCCollectAll() error {

return eng.gCCollectAll()

}

func (eng *sqLiteSystem) GetSQLEngine() sqlengine.SQLEngine {

return eng.sqlEngine

}

func (eng *sqLiteSystem) gCCollectAll() error {

obtainQuery := `

SELECT

'DELETE FROM "' || name || '" ; '

FROM

sqlite_master

where

type = 'table'

and

name not like '__iql__%%'

and

name NOT LIKE 'sqlite_%%'

and

name NOT IN (SELECT view_name FROM "__iql__.materialized_views")

and

name NOT IN (SELECT table_name FROM "__iql__.tables")

`

deleteQueryResultSet, err := eng.sqlEngine.Query(obtainQuery)

if err != nil {

return err

}

return eng.readExecGeneratedQueries(deleteQueryResultSet)

}

func (eng *sqLiteSystem) generateDropTableStatement(relationalTable relationaldto.RelationalTable) (string, error) {

s, err := relationalTable.GetName()

return fmt.Sprintf(`drop table if exists "%s"`, s), err

}

func (eng *sqLiteSystem) GCControlTablesPurge() error {

return eng.gcControlTablesPurge()

}

func (eng *sqLiteSystem) GenerateDDL(relationalTable relationaldto.RelationalTable, dropTable bool) ([]string, error) {

return eng.generateDDL(relationalTable, dropTable)

}

func (eng *sqLiteSystem) generateDDL(relationalTable relationaldto.RelationalTable, dropTable bool) ([]string, error) {

var colDefs, retVal []string

var rv strings.Builder

if dropTable {

dt, err := eng.generateDropTableStatement(relationalTable)

if err != nil {

return nil, err

}

retVal = append(retVal, dt)

}

tableName, err := relationalTable.GetName()

if err != nil {

return nil, err

}

rv.WriteString(fmt.Sprintf(`create table if not exists "%s" ( `, tableName))

colDefs = append(colDefs, fmt.Sprintf(`"iql_%s_id" INTEGER PRIMARY KEY AUTOINCREMENT`, tableName))

genIDColName := eng.controlAttributes.GetControlGenIDColumnName()

sessionIDColName := eng.controlAttributes.GetControlSsnIDColumnName()

txnIDColName := eng.controlAttributes.GetControlTxnIDColumnName()

maxTxnIDColName := eng.controlAttributes.GetControlMaxTxnColumnName()

insIDColName := eng.controlAttributes.GetControlInsIDColumnName()

lastUpdateColName := eng.controlAttributes.GetControlLatestUpdateColumnName()

insertEncodedColName := eng.controlAttributes.GetControlInsertEncodedIDColumnName()

gcStatusColName := eng.controlAttributes.GetControlGCStatusColumnName()

colDefs = append(colDefs, fmt.Sprintf(`"%s" INTEGER `, genIDColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" INTEGER `, sessionIDColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" INTEGER `, txnIDColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" INTEGER `, maxTxnIDColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" INTEGER `, insIDColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" TEXT `, insertEncodedColName))

colDefs = append(colDefs, fmt.Sprintf(`"%s" TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT CURRENT_TIMESTAMP `, lastUpdateColName)) //nolint:lll // this is a long line but it is a string

colDefs = append(colDefs, fmt.Sprintf(`"%s" SMALLINT NOT NULL DEFAULT %d `, gcStatusColName, constants.GCBlack))

for _, col := range relationalTable.GetColumns() {

var b strings.Builder

colName := col.GetName()

colType := col.GetType()

b.WriteString(`"` + colName + `" `)

b.WriteString(colType)

colDefs = append(colDefs, b.String())

}

rv.WriteString(strings.Join(colDefs, " , "))

rv.WriteString(" ) ")

retVal = append(retVal, rv.String())

retVal = append(retVal, fmt.Sprintf(`create index if not exists "idx_%s_%s" on "%s" ( "%s" ) `, strings.ReplaceAll(tableName, ".", "_"), genIDColName, tableName, genIDColName)) //nolint:lll // this is a long line but it is more readable this way

retVal = append(retVal, fmt.Sprintf(`create index if not exists "idx_%s_%s" on "%s" ( "%s" ) `, strings.ReplaceAll(tableName, ".", "_"), sessionIDColName, tableName, sessionIDColName)) //nolint:lll // this is a long line but it is more readable this way

retVal = append(retVal, fmt.Sprintf(`create index if not exists "idx_%s_%s" on "%s" ( "%s" ) `, strings.ReplaceAll(tableName, ".", "_"), txnIDColName, tableName, txnIDColName)) //nolint:lll // this is a long line but it is more readable this way

retVal = append(retVal, fmt.Sprintf(`create index if not exists "idx_%s_%s" on "%s" ( "%s" ) `, strings.ReplaceAll(tableName, ".", "_"), insIDColName, tableName, insIDColName)) //nolint:lll // this is a long line but it is more readable this way

rawViewDDL, err := eng.generateViewDDL(relationalTable)

if err != nil {

return nil, err

}

retVal = append(retVal, rawViewDDL...)

return retVal, nil

}

func (eng *sqLiteSystem) GetViewByName(viewName string) (internaldto.RelationDTO, bool) {

rv, ok := eng.getViewByName(viewName)

if !ok {

return nil, false

}

candidates, err := eng.getAwareViewsByName(viewName)

currentNode := rv

if err == nil {

for _, candidate := range candidates {

if rv.GetName() != candidate.GetName() {

currentNode = currentNode.WithNext(candidate)

}

}

}

return rv, ok

}

func (eng *sqLiteSystem) GetViewByNameAndParameters(

viewName string, params map[string]any) (internaldto.RelationDTO, bool) {

rv, err := eng.selectMatchingView(viewName, params)

if err != nil {

return nil, false

}

return rv, true

}

func (eng *sqLiteSystem) getViewByName(viewName string) (internaldto.RelationDTO, bool) {

q := `SELECT view_ddl, required_params FROM "__iql__.views" WHERE view_name = ? and deleted_dttm IS NULL`

row := eng.sqlEngine.QueryRow(q, viewName)

if row == nil {

return nil, false

}

var viewDDL, requiredParametersStr string

err := row.Scan(&viewDDL, &requiredParametersStr)

if err != nil {

return nil, false

}

paramSerDe := serde.NewStringArrayMapSerDe()

requiredParameters, serDeErr := paramSerDe.Deserialize(requiredParametersStr)

if serDeErr != nil {

return nil, false

}

rv := internaldto.NewViewDTO(viewName, viewDDL).WithRequiredParams(requiredParameters)

return rv, true

}

func (eng *sqLiteSystem) selectMatchingView(viewName string, params map[string]any) (internaldto.RelationDTO, error) {

candidates, err := eng.getAwareViewsByName(viewName)

if err != nil {

return nil, err

}

for _, candidate := range candidates {

if successfulCandidate, ok := candidate.MatchOnParams(params); ok {

return successfulCandidate, nil

}

}

return nil, fmt.Errorf("no matching view found for viewName = '%s'", viewName)

}

func (eng *sqLiteSystem) getAwareViewsByName(viewName string) ([]internaldto.RelationDTO, error) {

q := `SELECT view_name, view_ddl, required_params

FROM "__iql__.views" WHERE view_name LIKE ? and deleted_dttm IS NULL`

txn, err := eng.sqlEngine.GetTx()

if err != nil {

return nil, err

}

var rv []internaldto.RelationDTO

defer txn.Commit() //nolint:errcheck // TODO: establish pattern

rows, err := txn.Query(q, fmt.Sprintf(`%s%%`, viewName))

if err != nil {

return nil, err

}

defer rows.Close()

if rows.Err() != nil {

return nil, rows.Err()

}

var hasRow bool

for {

if !rows.Next() {

break

}

hasRow = true

var viewNameAware, viewDDL, requiredParametersStr string

err = rows.Scan(&viewNameAware, &viewDDL, &requiredParametersStr)

if err != nil {

return nil, err

}

paramSerDe := serde.NewStringArrayMapSerDe()

requiredParameters, serDeErr := paramSerDe.Deserialize(requiredParametersStr)

if serDeErr != nil {

return nil, serDeErr

}

viewDTO := internaldto.NewViewDTO(viewNameAware, viewDDL).WithRequiredParams(requiredParameters)

rv = append(rv, viewDTO)

}

if !hasRow {

return nil, fmt.Errorf("no views found for viewName = '%s'", viewName)

}

return rv, nil

}

func (eng *sqLiteSystem) DropView(viewName string) error {

_, err := eng.sqlEngine.Exec(`delete from "__iql__.views" where view_name = ?`, viewName)

return err

}

func (eng *sqLiteSystem) CreateView(

viewName string, rawDDL string, replaceAllowed bool, requiredParams []string) error {

return eng.createView(viewName, rawDDL, replaceAllowed, requiredParams)

}

func (eng *sqLiteSystem) createView(

viewName string, rawDDL string, replaceAllowed bool, requiredParams []string) error {

paramSerDe := serde.NewStringArrayMapSerDe()

requiredParamsString, serdeErr := paramSerDe.Serialize(requiredParams)

if serdeErr != nil {

return serdeErr

}

q := `

INSERT INTO "__iql__.views" (

view_name,

view_ddl,

required_params

)

VALUES (

?,

?,

?

)

`

if replaceAllowed {

q += `

ON CONFLICT(view_name)

DO

UPDATE SET view_ddl = EXCLUDED.view_ddl

`

}

_, err := eng.sqlEngine.Exec(q, viewName, rawDDL, requiredParamsString)

return err

}

func (eng *sqLiteSystem) generateViewDDL(relationalTable relationaldto.RelationalTable) ([]string, error) {

var colNames, retVal []string

var createViewBuilder strings.Builder

retVal = append(retVal, fmt.Sprintf(`drop view if exists "%s" ; `, relationalTable.GetBaseName()))

createViewBuilder.WriteString(fmt.Sprintf(`create view "%s" AS `, relationalTable.GetBaseName()))

for _, col := range relationalTable.GetColumns() {

var b strings.Builder

colName := col.DelimitedSelectionString(`"`)

b.WriteString(colName)

colNames = append(colNames, b.String())

}

tableName, err := relationalTable.GetName()

if err != nil {

return nil, err

}

createViewBuilder.WriteString(fmt.Sprintf(`select %s from "%s" ;`, strings.Join(colNames, ", "), tableName))

retVal = append(retVal, createViewBuilder.String())

return retVal, nil

}

//nolint:unparam,revive // future proof

func (eng *sqLiteSystem) CreateMaterializedView(

relationName string,

colz []typing.RelationalColumn,

rawDDL string,

replaceAllowed bool,

selectQuery string,

varargs ...any,

) error {

return eng.runMaterializedViewCreate(

relationName,

colz,

rawDDL,

replaceAllowed,

selectQuery,

varargs...,

)

}

//nolint:errcheck,revive,staticcheck // TODO: establish pattern

func (eng *sqLiteSystem) RefreshMaterializedView(naiveViewName string,

colz []typing.RelationalColumn,

selectQuery string,

varargs ...any) error {

fullyQualifiedRelationName := eng.getFullyQualifiedRelationName(naiveViewName)

//nolint:gosec // no viable alternative

deleteQuery := fmt.Sprintf(`

DELETE FROM "%s"`,

fullyQualifiedRelationName,

)

txn, err := eng.sqlEngine.GetTx()

if err != nil {

return err

}

_, err = txn.Exec(deleteQuery)

if err != nil {

txn.Rollback()

return err

}

// TODO: check colz against DTO

relationDTO, relationDTOok := eng.getMaterializedViewByName(naiveViewName, txn)

if !relationDTOok {

if len(relationDTO.GetColumns()) == 0 {

}

// no need to rollbak; assumed already done

return fmt.Errorf("cannot refresh materialized view = '%s': not found", naiveViewName)

}

insertQuery := eng.generateTableInsertDMLFromViewSelect(fullyQualifiedRelationName, selectQuery, colz)

_, err = txn.Exec(insertQuery, varargs...)

if err != nil {

txn.Rollback()

return err

}

commitErr := txn.Commit()

return commitErr

}

//nolint:errcheck,revive,staticcheck // TODO: establish pattern

func (eng *sqLiteSystem) InsertIntoPhysicalTable(naiveTableName string,

columnsString string,

selectQuery string,

varargs ...any) error {

txn, err := eng.sqlEngine.GetTx()

if err != nil {

return err

}

fullyQualifiedRelationName := eng.getFullyQualifiedRelationName(naiveTableName)

// TODO: check colz against supplied columns

relationDTO, relationDTOok := eng.getTableByName(naiveTableName, txn)

if !relationDTOok {

if len(relationDTO.GetColumns()) == 0 {

}

// no need to rollbak; assumed already done

return fmt.Errorf("cannot refresh materialized view = '%s': not found", fullyQualifiedRelationName)

}

//nolint:gosec // no viable alternative

insertQuery := fmt.Sprintf(`INSERT INTO "%s" %s %s`, fullyQualifiedRelationName, columnsString, selectQuery)

_, err = txn.Exec(insertQuery, varargs...)

if err != nil {

txn.Rollback()

return err

}

commitErr := txn.Commit()

return commitErr

}

//nolint:errcheck // TODO: establish pattern

func (eng *sqLiteSystem) DropMaterializedView(naiveViewName string) error {

fullyQualifiedRelationName := eng.getFullyQualifiedRelationName(naiveViewName)

dropRefQuery := `

DELETE FROM "__iql__.materialized_views"

WHERE view_name = ?

`

dropColsQuery := `

DELETE

FROM

"__iql__.materialized_views.columns"

WHERE

view_name = ?

`

dropTableQuery := fmt.Sprintf(`

DROP TABLE IF EXISTS "%s"

`, fullyQualifiedRelationName)

tx, err := eng.sqlEngine.GetTx()

if err != nil {

return err

}

_, err = tx.Exec(dropRefQuery, fullyQualifiedRelationName)

if err != nil {

tx.Rollback()

return err

}

_, err = tx.Exec(dropColsQuery, fullyQualifiedRelationName)

if err != nil {

tx.Rollback()

return err

}

_, err = tx.Exec(dropTableQuery)

if err != nil {

tx.Rollback()

return err

}

commitErr := tx.Commit()

return commitErr

}

//nolint:errcheck // TODO: establish pattern

func (eng *sqLiteSystem) GetMaterializedViewByName(viewName string) (internaldto.RelationDTO, bool) {

txn, err := eng.sqlEngine.GetTx()

if err != nil {

return nil, false

}

rv, ok := eng.getMaterializedViewByName(viewName, txn)

txn.Commit()

return rv, ok

}

func (eng *sqLiteSystem) IsRelationExported(relationName string) bool {

if eng.exportNamespace == "" {

return false

}

matches, _ := regexp.MatchString(fmt.Sprintf(`^%s.*$`, eng.exportNamespace), relationName)

return matches

}

//nolint:errcheck,gosec // TODO: establish pattern

func (eng *sqLiteSystem) getMaterializedViewByName(naiveViewName string, txn *sql.Tx) (internaldto.RelationDTO, bool) {

fullyQualifiedRelationName := eng.getFullyQualifiedRelationName(naiveViewName)

q := `SELECT view_ddl FROM "__iql__.materialized_views" WHERE view_name = ? and deleted_dttm IS NULL`

colQuery := `

SELECT

column_name

,column_type

,"oid"

,column_width

,column_precision

FROM

"__iql__.materialized_views.columns"

WHERE

view_name = ?

ORDER BY ordinal_position ASC

`

// txn, txnErr := eng.sqlEngine.GetTx()

// if txnErr != nil {

// return nil, false

// }

row := txn.QueryRow(q, fullyQualifiedRelationName)

if row == nil {

txn.Rollback()

return nil, false

}

var viewDDL string

err := row.Scan(&viewDDL)

if err != nil {

txn.Rollback()

return nil, false

}

rv := internaldto.NewMaterializedViewDTO(fullyQualifiedRelationName, viewDDL, eng.exportNamespace)

rows, err := txn.Query(colQuery, fullyQualifiedRelationName)

if err != nil || rows == nil || rows.Err() != nil {

txn.Rollback()

return nil, false

}

defer rows.Close()

hasRow := false

var columns []typing.RelationalColumn

for {

if !rows.Next() {

break

}

hasRow = true

var columnName, columnType string

var oID, colWidth, colPrecision int

err = rows.Scan(&columnName, &columnType, &oID, &colWidth, &colPrecision)

if err != nil {

txn.Rollback()

return nil, false

}

relationalColumn := typing.NewRelationalColumn(

columnName,

columnType).WithWidth(colWidth).WithOID(oid.Oid(oID))

columns = append(columns, relationalColumn)

}

rv.SetColumns(columns)

if !hasRow {

txn.Rollback()

return nil, false

}

return rv, true

}

//nolint:errcheck // TODO: establish pattern

func (eng *sqLiteSystem) GetPhysicalTableByName(

tableName string) (internaldto.RelationDTO, bool) {

txn, err := eng.sqlEngine.GetTx()

if err != nil {

return nil, false

}

rv, ok := eng.getTableByName(tableName, txn)

txn.Commit()

return rv, ok

}

// TODO: implement temp tables

//

//nolint:errcheck,gosec // TODO: establish pattern

func (eng *sqLiteSystem) getTableByName(

naiveTableName string,

txn *sql.Tx,

) (internaldto.RelationDTO, bool) {

fullyQualifiedTableName := eng.getFullyQualifiedRelationName(naiveTableName)

q := `SELECT table_ddl FROM "__iql__.tables" WHERE table_name = ? and deleted_dttm IS NULL`

colQuery := `

SELECT

column_name

,column_type

,"oid"

,column_width

,column_precision

FROM

"__iql__.tables.columns"

WHERE

table_name = ?

ORDER BY ordinal_position ASC

`

row := txn.QueryRow(q, fullyQualifiedTableName)

if row == nil {

txn.Rollback()

return nil, false

}

var viewDDL string

err := row.Scan(&viewDDL)

if err != nil {

txn.Rollback()

return nil, false

}

rv := internaldto.NewPhysicalTableDTO(fullyQualifiedTableName, viewDDL, eng.exportNamespace)

rows, err := txn.Query(colQuery, fullyQualifiedTableName)

if err != nil || rows == nil || rows.Err() != nil {

txn.Rollback()

return nil, false

}

defer rows.Close()

hasRow := false

var columns []typing.RelationalColumn

for {

if !rows.Next() {

break

}

hasRow = true

var columnName, columnType string

var oID, colWidth, colPrecision int

err = rows.Scan(&columnName, &columnType, &oID, &colWidth, &colPrecision)

if err != nil {

txn.Rollback()

return nil, false

}

relationalColumn := typing.NewRelationalColumn(

columnName,

columnType).WithWidth(colWidth).WithOID(oid.Oid(oID))

columns = append(columns, relationalColumn)

}

rv.SetColumns(columns)

if !hasRow {

txn.Rollback()

return nil, false

}

return rv, true

}

// TODO: implement temp table drop