Merge pull request #28 from gigapi/feat/s3

Feat: S3 support

Author: akvlad

go.mod

@@ -4,7 +4,7 @@ go 1.24.2
 
 require (
 	github.com/apache/arrow/go/v14 v14.0.2
-	github.com/gigapi/gigapi-config v0.0.8
+	github.com/gigapi/gigapi-config v0.0.9
 	github.com/gigapi/gigapi/v2 v2.0.13
 	github.com/gigapi/metadata v0.0.4
 	github.com/marcboeker/go-duckdb/v2 v2.2.0

go.sum

@@ -38,8 +38,8 @@ github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHk
 github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
 github.com/fsnotify/fsnotify v1.7.0 h1:8JEhPFa5W2WU7YfeZzPNqzMP6Lwt7L2715Ggo0nosvA=
 github.com/fsnotify/fsnotify v1.7.0/go.mod h1:40Bi/Hjc2AVfZrqy+aj+yEI+/bRxZnMJyTJwOpGvigM=
-github.com/gigapi/gigapi-config v0.0.8 h1:fyOwS3by+qMlHLthmIgX1gXEchH2PUH2l4x/y9sRyOg=
-github.com/gigapi/gigapi-config v0.0.8/go.mod h1:/hD+d1odWyElSP9++ZPLULXthWHXvx0shNypdbZDYA8=
+github.com/gigapi/gigapi-config v0.0.9 h1:7VWmMJb2dKGDReasOgu45bzOmrfj/WCiBcO9g4MNgXw=
+github.com/gigapi/gigapi-config v0.0.9/go.mod h1:/hD+d1odWyElSP9++ZPLULXthWHXvx0shNypdbZDYA8=
 github.com/gigapi/gigapi/v2 v2.0.13 h1:m4w2Sx+AQZl+SPS+5oCtCe7G1888/2nF6m+d1vsiHlQ=
 github.com/gigapi/gigapi/v2 v2.0.13/go.mod h1:xoh5uz+EQCK43cI08wMJlyhAZ1368OOPlJjmOr1TGis=
 github.com/gigapi/metadata v0.0.4 h1:kQdreAUyeguIrwSlPukzZ36Y9HlKVhcaBgrg8crK1tY=

querier/layerDesc.go

@@ -0,0 +1,67 @@
+package querier
+
+import (
+	"fmt"
+	"github.com/gigapi/gigapi-config/config"
+	"net/url"
+	"strings"
+)
+
+type querierLayerDesc struct {
+	layer    config.LayersConfiguration
+	path     string
+	hostname string
+	key      string
+	secret   string
+	bucket   string
+	secure   bool
+	urlStyle string
+}
+
+func getQuerierLayerDesc(layer config.LayersConfiguration) (querierLayerDesc, error) {
+	switch layer.Type {
+	case "fs":
+		return getQuerierLayerDescFs(layer)
+	case "s3":
+		return getQuerierLayerDescS3(layer)
+	}
+	return querierLayerDesc{}, fmt.Errorf("Unsupported layer type: %s", layer.Type)
+}
+
+func getQuerierLayerDescFs(layer config.LayersConfiguration) (querierLayerDesc, error) {
+	return querierLayerDesc{
+		layer: layer,
+		path:  strings.TrimPrefix(layer.URL, "file://"),
+	}, nil
+}
+
+func getQuerierLayerDescS3(layer config.LayersConfiguration) (querierLayerDesc, error) {
+	s3Url, err := url.Parse(layer.URL)
+	if err != nil {
+		return querierLayerDesc{}, err
+	}
+	if s3Url.Scheme != "s3" {
+		return querierLayerDesc{}, fmt.Errorf("Invalid S3 URL: %s", layer.URL)
+	}
+
+	pathParts := strings.SplitN(s3Url.Path, "/", 2)
+
+	res := querierLayerDesc{
+		layer:    layer,
+		hostname: s3Url.Host,
+		bucket:   pathParts[0],
+		path:     pathParts[1],
+		secure:   s3Url.Query().Get("secure") != "false",
+		key:      layer.Auth.Key,
+		secret:   layer.Auth.Secret,
+	}
+	if s3Url.User != nil {
+		res.key = s3Url.User.Username()
+		res.secret, _ = s3Url.User.Password()
+	}
+	res.urlStyle = "vhost"
+	if s3Url.Query().Get("url-style") == "path" {
+		res.urlStyle = "path"
+	}
+	return res, nil
+}

querier/queryClient.go

@@ -30,6 +30,8 @@ type QueryClient struct {
 	DataDir          string
 	DB               *sql.DB
 	DefaultTimeRange int64 // 10 minutes in nanoseconds
+
+	layers map[string]querierLayerDesc
 }
 
 // NewQueryClient creates a new QueryClient
@@ -48,6 +50,16 @@ func (q *QueryClient) Initialize() error {
 		return fmt.Errorf("failed to initialize DuckDB: %v", err)
 	}
 	q.DB = db
+	q.layers = map[string]querierLayerDesc{}
+
+	for _, l := range config.Config.Gigapi.Layers {
+		desc, err := getQuerierLayerDesc(l)
+		if err != nil {
+			return fmt.Errorf("failed to initialize layer %s: %v", l.Name, err)
+		}
+		q.layers[l.Name] = desc
+	}
+
 	return nil
 }
 
@@ -422,7 +434,7 @@ func getTableIndex(table *shared.Table) (metadata.TableIndex, error) {
 
 // Find relevant parquet files based on time range
 func (q *QueryClient) FindRelevantFiles(ctx context.Context, dbName, measurement string,
-	timeRange TimeRange) ([]string, error) {
+	timeRange TimeRange) ([]*metadata.IndexEntry, error) {
 	// If no time range specified, get all files
 	var relevantFiles []string
 	// log.Printf("Getting relevant files for %s.%s within time range %v to %v", dbName, measurement,
@@ -451,19 +463,13 @@ func (q *QueryClient) FindRelevantFiles(ctx context.Context, dbName, measurement
 	}
 
 	ies, err := idx.GetQuerier().Query(opts)
-	layersMap := make(map[string]string)
-	for _, l := range config.Config.Gigapi.Layers {
-		layersMap[l.Name] = strings.TrimPrefix(l.URL, "file://")
-	}
-	for _, ie := range ies {
-		relevantFiles = append(relevantFiles, filepath.Join(layersMap[ie.Layer], dbName, measurement, "data", ie.Path))
+	if err != nil {
+		return nil, err
 	}
-
-	if len(relevantFiles) == 0 {
+	if len(ies) == 0 {
 		log.Printf("No files found in any directory for %s.%s", dbName, measurement)
 	}
-
-	return relevantFiles, nil
+	return ies, err
 }
 
 // Find all files for a measurement
@@ -660,6 +666,49 @@ func (c *QueryClient) getDBIndex(ctx context.Context) (metadata.DBIndex, error)
 	return nil, fmt.Errorf("unsupported metadata type: %s", config.Config.Gigapi.Metadata.Type)
 }
 
+func (c *QueryClient) buildFilesList(db string, table string, files []*metadata.IndexEntry) ([]string, error) {
+	var fileList []string
+	s3Layers := map[string]bool{}
+
+	for _, f := range files {
+		l, ok := c.layers[f.Layer]
+		if !ok {
+			return nil, fmt.Errorf("layer %s not found", f.Layer)
+		}
+		switch l.layer.Type {
+		case "fs":
+			fileList = append(fileList, filepath.Join(l.path, db, table, "data", f.Path))
+		case "s3":
+			_path := l.path
+			if _path != "" {
+				_path += "/"
+			}
+			path := fmt.Sprintf("s3://%s%s/%s/%s", _path, db, table, f.Path)
+			fileList = append(fileList, path)
+			s3Layers[l.layer.Name] = true
+		}
+	}
+	for lName, _ := range s3Layers {
+		sanitizedLName := lName
+		sanitizedLName = regexp.MustCompile("[^a-zA-Z0-9_]").ReplaceAllString(sanitizedLName, "_")
+		secretName := fmt.Sprintf("secret_%s", sanitizedLName)
+		l := c.layers[lName]
+		_, err := c.DB.Exec(fmt.Sprintf(`CREATE OR REPLACE SECRET %s (
+    TYPE s3,
+    USE_SSL %t,
+    KEY_ID %s,
+    SECRET %s,
+	ENDPOINT '%s',
+	SCOPE 's3://%s',
+    URL_STYLE '%s'
+);`, secretName, l.secure, l.key, l.secret, l.hostname, l.bucket, l.urlStyle))
+		if err != nil {
+			return nil, fmt.Errorf("failed to create secret: %v", err)
+		}
+	}
+	return fileList, nil
+}
+
 // Query executes a query against the database
 func (c *QueryClient) Query(ctx context.Context, query, dbName string) ([]map[string]interface{}, error) {
 	// Ensure we have a context
@@ -766,9 +815,13 @@ func (c *QueryClient) Query(ctx context.Context, query, dbName string) ([]map[st
 	}
 
 	// Find relevant files
-	files, err := c.FindRelevantFiles(ctx, parsed.DbName, parsed.Measurement, parsed.TimeRange)
-	if err != nil || len(files) == 0 {
-		return nil, fmt.Errorf("no relevant files found for query")
+	entries, err := c.FindRelevantFiles(ctx, parsed.DbName, parsed.Measurement, parsed.TimeRange)
+	if err != nil {
+		return nil, err
+	}
+	files, err := c.buildFilesList(parsed.DbName, parsed.Measurement, entries)
+	if err != nil {
+		return nil, err
 	}
 
 	start := time.Now()