这是indexloc提供的服务，不要输入任何密码
Skip to content

Ducklake migration #552

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 25 commits into from
Sep 16, 2025
Merged

Ducklake migration #552

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
25 commits
Select commit Hold shift + click to select a range
913db8f
initial - migration
pskrbasu Sep 3, 2025
8bbe727
cancellation and run for every command
pskrbasu Sep 7, 2025
acee926
cancellation and print summary
pskrbasu Sep 7, 2025
10d3164
spinner and summary
pskrbasu Sep 9, 2025
4c97063
changes with rebase
pskrbasu Sep 9, 2025
ebb01c4
ducklake option - write
pskrbasu Sep 9, 2025
0dbc44f
updated status messages
pskrbasu Sep 9, 2025
8bb557a
refactor
pskrbasu Sep 9, 2025
be094c7
Enhance migration process by adding parquet file counting and updatin…
pskrbasu Sep 9, 2025
8499cb4
cleanup
pskrbasu Sep 15, 2025
a5dd813
lint
pskrbasu Sep 15, 2025
73f626b
lint
pskrbasu Sep 15, 2025
8c572aa
cleanup
pskrbasu Sep 15, 2025
d673467
initial revision
pskrbasu Sep 15, 2025
62d6cd8
revision
pskrbasu Sep 15, 2025
a025370
cleanup
pskrbasu Sep 15, 2025
47ca1b9
use contexthelpers.StartCancelHandler
pskrbasu Sep 16, 2025
7fd2cbf
add better comment (suppress cobra logs)
pskrbasu Sep 16, 2025
5a19d75
add better comments
pskrbasu Sep 16, 2025
4f5e8ad
tidy up context usage, cancellation handling and general structure
kaidaguerre Sep 16, 2025
0d2903f
minor changes
pskrbasu Sep 16, 2025
beecf73
pf v2.7.0-rc.1
pskrbasu Sep 16, 2025
c35f831
refactor error handling in format commands to use a consistent exit code
Priyanka-Chatterjee-2000 Sep 16, 2025
885a2ad
remove cancelhandlers from cmd; use from prerunhooks
pskrbasu Sep 16, 2025
a72ac0d
sdk v0.9.3
pskrbasu Sep 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions cmd/collect.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/modconfig"
"github.com/turbot/pipe-fittings/v2/parse"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
Expand Down Expand Up @@ -61,8 +60,9 @@ Every time you run tailpipe collect, Tailpipe refreshes its views over all colle
}

func runCollectCmd(cmd *cobra.Command, args []string) {
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
ctx, cancel := context.WithCancel(ctx) //nolint:govet // cancel is needed for the doCollect func

var err error
defer func() {
Expand All @@ -83,7 +83,7 @@ func runCollectCmd(cmd *cobra.Command, args []string) {
// if diagnostic mode is set, print out config and return
if _, ok := os.LookupEnv(constants.EnvConfigDump); ok {
localcmdconfig.DisplayConfig()
return
return //nolint:govet // this is explicitly used in tests
}

err = doCollect(ctx, cancel, args)
Expand Down
5 changes: 2 additions & 3 deletions cmd/compact.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
"github.com/turbot/tailpipe/internal/config"
"github.com/turbot/tailpipe/internal/constants"
Expand All @@ -38,8 +37,8 @@ func compactCmd() *cobra.Command {

func runCompactCmd(cmd *cobra.Command, args []string) {
var err error
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()

defer func() {
if r := recover(); r != nil {
Expand Down
8 changes: 3 additions & 5 deletions cmd/connect.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,23 +10,21 @@ import (
"strings"
"time"

"golang.org/x/exp/maps"

"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/thediveo/enumflag/v2"
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
"github.com/turbot/pipe-fittings/v2/connection"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
pfilepaths "github.com/turbot/pipe-fittings/v2/filepaths"
"github.com/turbot/pipe-fittings/v2/parse"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
"github.com/turbot/tailpipe/internal/config"
"github.com/turbot/tailpipe/internal/constants"
"github.com/turbot/tailpipe/internal/database"
error_helpers "github.com/turbot/tailpipe/internal/error_helpers"
"golang.org/x/exp/maps"
)

// variable used to assign the output mode flag
Expand Down Expand Up @@ -96,8 +94,8 @@ The generated script can be used with DuckDB:
func runConnectCmd(cmd *cobra.Command, _ []string) {
var err error
var initFilePath string
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()

defer func() {
if r := recover(); r != nil {
Expand Down
20 changes: 8 additions & 12 deletions cmd/format.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package cmd

import (
"context"
"fmt"
"os"
"strings"
Expand All @@ -11,7 +10,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/printers"
"github.com/turbot/pipe-fittings/v2/utils"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
Expand Down Expand Up @@ -69,9 +67,8 @@ func formatListCmd() *cobra.Command {
}

func runFormatListCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runFormatListCmd start")
var err error
defer func() {
Expand All @@ -86,7 +83,7 @@ func runFormatListCmd(cmd *cobra.Command, args []string) {
} else {
error_helpers.ShowError(ctx, err)
}
setExitCodeForFormatError(err,1)
setExitCodeForFormatError(err)
}
}()

Expand Down Expand Up @@ -133,9 +130,8 @@ func formatShowCmd() *cobra.Command {
}

func runFormatShowCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runFormatShowCmd start")
var err error
defer func() {
Expand All @@ -150,7 +146,7 @@ func runFormatShowCmd(cmd *cobra.Command, args []string) {
} else {
error_helpers.ShowError(ctx, err)
}
setExitCodeForFormatError(err, 1)
setExitCodeForFormatError(err)
}
}()

Expand All @@ -172,7 +168,7 @@ func runFormatShowCmd(cmd *cobra.Command, args []string) {
}
}

func setExitCodeForFormatError(err error, nonCancelCode int) {
func setExitCodeForFormatError(err error) {
// set exit code only if an error occurred and no exit code is already set
if exitCode != 0 || err == nil {
return
Expand All @@ -183,5 +179,5 @@ func setExitCodeForFormatError(err error, nonCancelCode int) {
return
}
// no dedicated format exit code exists yet; use generic nonzero failure
exitCode = nonCancelCode
exitCode = 1
}
23 changes: 10 additions & 13 deletions cmd/partition.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/printers"
"github.com/turbot/pipe-fittings/v2/statushooks"
"github.com/turbot/pipe-fittings/v2/utils"
Expand Down Expand Up @@ -76,9 +75,8 @@ func partitionListCmd() *cobra.Command {
}

func runPartitionListCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runPartitionListCmd start")
var err error
defer func() {
Expand All @@ -103,7 +101,6 @@ func runPartitionListCmd(cmd *cobra.Command, args []string) {
return
}

// open a readonly db connection
db, err := database.NewDuckDb(database.WithDuckLakeReadonly())
error_helpers.FailOnError(err)
defer db.Close()
Expand Down Expand Up @@ -146,10 +143,10 @@ func partitionShowCmd() *cobra.Command {
}

func runPartitionShowCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()
utils.LogTime("runPartitionShowCmd start")
var err error
defer func() {
Expand Down Expand Up @@ -233,10 +230,10 @@ func partitionDeleteCmd() *cobra.Command {
}

func runPartitionDeleteCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()
var err error
defer func() {
if r := recover(); r != nil {
Expand Down
42 changes: 20 additions & 22 deletions cmd/plugin.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/filepaths"
"github.com/turbot/pipe-fittings/v2/installationstate"
pociinstaller "github.com/turbot/pipe-fittings/v2/ociinstaller"
Expand Down Expand Up @@ -238,10 +237,10 @@ var pluginInstallSteps = []string{
}

func runPluginInstallCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()
utils.LogTime("runPluginInstallCmd install")
var err error
defer func() {
Expand Down Expand Up @@ -386,10 +385,10 @@ func doPluginInstall(ctx context.Context, bar *uiprogress.Bar, pluginName string
}

func runPluginUpdateCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()
utils.LogTime("runPluginUpdateCmd start")
var err error
defer func() {
Expand Down Expand Up @@ -667,10 +666,10 @@ func installPlugin(ctx context.Context, resolvedPlugin pplugin.ResolvedPluginVer
}

func runPluginUninstallCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()

utils.LogTime("runPluginUninstallCmd uninstall")
var err error
Expand Down Expand Up @@ -768,9 +767,8 @@ func resolveUpdatePluginsFromArgs(args []string) ([]string, error) {
}

func runPluginListCmd(cmd *cobra.Command, _ []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()

utils.LogTime("runPluginListCmd list")

Expand Down Expand Up @@ -822,18 +820,18 @@ func runPluginListCmd(cmd *cobra.Command, _ []string) {
}

func runPluginShowCmd(cmd *cobra.Command, args []string) {
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()

// we expect 1 argument, the plugin name
if len(args) != 1 {
error_helpers.ShowError(cmd.Context(), fmt.Errorf("you need to provide the name of a plugin"))
error_helpers.ShowError(ctx, fmt.Errorf("you need to provide the name of a plugin"))
exitCode = pconstants.ExitCodeInsufficientOrWrongInputs
return
}

//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)

utils.LogTime("runPluginShowCmd start")

// Clean up plugin temporary directories from previous crashes/interrupted installations
Expand Down
17 changes: 7 additions & 10 deletions cmd/source.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package cmd

import (
"context"
"fmt"
"os"
"strings"
Expand All @@ -11,13 +10,12 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/error_helpers"
"github.com/turbot/pipe-fittings/v2/printers"
"github.com/turbot/pipe-fittings/v2/utils"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
"github.com/turbot/tailpipe/internal/constants"
"github.com/turbot/tailpipe/internal/display"
error_helpers "github.com/turbot/tailpipe/internal/error_helpers"
)

func sourceCmd() *cobra.Command {
Expand Down Expand Up @@ -66,9 +64,8 @@ func sourceListCmd() *cobra.Command {
}

func runSourceListCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runSourceListCmd start")
var err error
defer func() {
Expand Down Expand Up @@ -130,10 +127,10 @@ func sourceShowCmd() *cobra.Command {
}

func runSourceShowCmd(cmd *cobra.Command, args []string) {
//setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
//TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a cancellation error. Cancellation won't work right now
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
// TODO: https://github.com/turbot/tailpipe/issues/563 none of the functions called in this command will return a
// cancellation error. Cancellation won't work right now
ctx := cmd.Context()
utils.LogTime("runSourceShowCmd start")
var err error
defer func() {
Expand Down
12 changes: 4 additions & 8 deletions cmd/table.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
package cmd

import (
"context"
"fmt"
"os"
"strings"
Expand All @@ -11,7 +10,6 @@ import (
"github.com/turbot/go-kit/helpers"
"github.com/turbot/pipe-fittings/v2/cmdconfig"
pconstants "github.com/turbot/pipe-fittings/v2/constants"
"github.com/turbot/pipe-fittings/v2/contexthelpers"
"github.com/turbot/pipe-fittings/v2/printers"
"github.com/turbot/pipe-fittings/v2/utils"
localcmdconfig "github.com/turbot/tailpipe/internal/cmdconfig"
Expand Down Expand Up @@ -68,9 +66,8 @@ func tableListCmd() *cobra.Command {
}

func runTableListCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runSourceListCmd start")
var err error
defer func() {
Expand Down Expand Up @@ -138,9 +135,8 @@ func tableShowCmd() *cobra.Command {
}

func runTableShowCmd(cmd *cobra.Command, args []string) {
// setup a cancel context and start cancel handler
ctx, cancel := context.WithCancel(cmd.Context())
contexthelpers.StartCancelHandler(cancel)
// use the signal-aware/cancelable context created upstream in preRunHook
ctx := cmd.Context()
utils.LogTime("runTableShowCmd start")
var err error
defer func() {
Expand Down
Loading