Merge branch 'master' of github.com:molecula/featurebase into clustertest-retry-coverage-v2

.circleci/config.yml

@@ -1,305 +0,0 @@
-# version: 2.1
-
-# executors:
-#   golang:
-#     parameters:
-#       version:
-#         type: string
-#         default: "1.15.8"
-#       resource_class:
-#         type: string
-#         default: medium
-#     docker:
-#       - image: circleci/golang:<< parameters.version >>
-#     resource_class: << parameters.resource_class >>
-#     working_directory: /go/src/github.com/molecula/featurebase
-
-# commands:
-#   add-github-auth:
-#     steps:
-#       - run: git config --global url."https://${GITHUB_USER}:${GITHUB_PERSONAL_ACCESS_TOKEN}@github.com/".insteadOf "https://github.com/"
-#       - run: git config --global url."https://${GITHUB_USER}:${GITHUB_PERSONAL_ACCESS_TOKEN}@github.com/".insteadOf "git@github.com:"
-#   restore-mod-cache:
-#     steps:
-#       - restore_cache:
-#           key: mod-cache-{{ checksum "go.sum" }}
-#   save-mod-cache:
-#     steps:
-#       - save_cache:
-#           key: mod-cache-{{ checksum "go.sum" }}
-#           paths:
-#             - /go/pkg/mod/
-#   checkout-plus:
-#     steps:
-#       - add-github-auth
-#       - checkout
-#       - restore-mod-cache
-#   skip-if-root-unchanged:
-#     description: "skips the parent job if the PR includes no changes to featurebase"
-#     steps:
-#       - run: |
-#           ROOT_CHANGED_FILES="$(git diff --name-only HEAD $(git merge-base master HEAD) | grep -v '^lattice/')" || true
-#           echo "ROOT_CHANGED_FILES = $ROOT_CHANGED_FILES"
-#           if [ -z "$ROOT_CHANGED_FILES" ] ; then
-#             echo "halting step"
-#             circleci step halt
-#           fi
-#   skip-if-lattice-unchanged:
-#     description: "skips the parent job if the PR includes no changes to lattice"
-#     steps:
-#       - run: |
-#           LATTICE_CHANGED_FILES="$(git diff --name-only HEAD $(git merge-base master HEAD) | grep '^lattice/')" || true
-#           echo "LATTICE_CHANGED_FILES = $LATTICE_CHANGED_FILES"
-#           if [ -z "$LATTICE_CHANGED_FILES" ] ; then
-#             echo "halting step"
-#             circleci step halt
-#           fi
-
-# jobs:
-#   setup:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - run: go mod download
-#       - save-mod-cache
-#   linter:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - run: curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sudo sh -s -- -b /usr/local/bin v1.31.0
-#       - run: make golangci-lint
-#   go-mod-tidy:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - run: go mod tidy
-#       - run: git diff --exit-code -- go.mod go.sum
-#   check-changelog-label:
-#     executor:
-#       name: golang
-#     steps:
-#       - run: '[[ -n $CIRCLE_PULL_REQUEST ]] || circleci step halt || true' # Skip if this is not a pull request
-#       - run: curl https://$GITHUB_USER:$GITHUB_PERSONAL_ACCESS_TOKEN@api.github.com/repos/molecula/featurebase/pulls/$(basename $CIRCLE_PULL_REQUEST) | jq "[.labels[] | .name | startswith(\"changelog\")] | any" -e
-#   test-build-arm:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - run: make build GOOS=linux GOARCH=arm GOARM=5
-#       - run: make build GOOS=linux GOARCH=arm GOARM=6
-#       - run: make build GOOS=linux GOARCH=arm GOARM=7
-#       - run: make build GOOS=linux GOARCH=arm64
-#   test:
-#     parameters:
-#       resource_class:
-#         type: string
-#         default: medium
-#       golang_version:
-#         type: string
-#         default: "1.15.8"
-#       shard_width:
-#         type: string
-#         default: "20"
-#       test_make_target:
-#         type: string
-#         default: "test"
-#       test_flags:
-#         type: string
-#         default: ""
-#       goarch:
-#         type: string
-#         default: amd64
-#     executor:
-#       name: golang
-#       version: << parameters.golang_version >>
-#       resource_class: << parameters.resource_class >>
-#     environment:
-#       TMPDIR: /mnt/ramdisk
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - run: sudo apt-get update --allow-releaseinfo-change -y
-#       - run: sudo apt-get install lsof
-#       - run:
-#           command: make << parameters.test_make_target >> SHARD_WIDTH=<< parameters.shard_width >> GOARCH=<< parameters.goarch >>
-#           no_output_timeout: 30m
-#   test-external-lookup:
-#     docker:
-#       - image: circleci/golang:1.15.8
-#       - image: circleci/postgres:13.2-ram
-#         environment:
-#           POSTGRES_PASSWORD=password
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - run: sudo apt-get update --allow-releaseinfo-change -y
-#       - run: sudo apt-get install postgresql-client
-#       - run: (for i in `seq 1 20`; do pg_isready -h localhost && exit 0 || sleep 1; done; exit 1)
-#       - run:
-#           command: make test-external-lookup EXTERNAL_LOOKUP_DSN=postgresql://postgres:password@localhost/circle_test?sslmode=disable
-#           no_output_timeout: 30m
-#   cluster-tests:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - skip-if-root-unchanged
-#       - setup_remote_docker
-#       - run: make clustertests
-#   release:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - attach_workspace:
-#           at: .
-#       - setup_remote_docker:
-#           version: 19.03.13 # see https://support.circleci.com/hc/en-us/articles/360050934711
-#       - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin
-#       - run: make docker-release
-#       - store_artifacts:
-#           path: build
-#       - persist_to_workspace:
-#           root: .
-#           paths: build
-#   publish_release:
-#     executor:
-#       name: golang
-#     steps:
-#       - attach_workspace:
-#           at: .
-#       - run: go get github.com/tcnksm/ghr
-#       - run: ghr -t ${GITHUB_PERSONAL_ACCESS_TOKEN} -u ${CIRCLE_PROJECT_USERNAME} -r ${CIRCLE_PROJECT_REPONAME} -c ${CIRCLE_SHA1} -delete ${CIRCLE_TAG} ./build/
-#   docker-build:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - setup_remote_docker:
-#           version: 19.03.13 # see https://support.circleci.com/hc/en-us/articles/360050934711
-#       - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin
-#       - run: make docker GO_VERSION=1.15.8
-#       - run: docker run featurebase:$(git describe --tags) help
-#   dockerhub-upload-unstable:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - setup_remote_docker:
-#           version: 19.03.13 # see https://support.circleci.com/hc/en-us/articles/360050934711
-#       - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin
-#       - run: make docker
-#       - run: docker run featurebase:$(git describe --tags) help
-#       - run: make docker-tag-push DOCKER_TARGET=moleculacorp/featurebase:<< pipeline.git.branch >>
-#   dockerhub-upload-stable:
-#     executor:
-#       name: golang
-#     steps:
-#       - checkout-plus
-#       - setup_remote_docker:
-#           version: 19.03.13 # see https://support.circleci.com/hc/en-us/articles/360050934711
-#       - run: echo -n $DOCKER_PASS | docker login -u $DOCKER_USER --password-stdin
-#       - run: make docker
-#       - run: docker run featurebase:$(git describe --tags) help
-#       - run: make docker-tag-push DOCKER_TARGET=moleculacorp/featurebase:<< pipeline.git.tag >>
-#       - run: make docker-tag-push DOCKER_TARGET=moleculacorp/featurebase:latest
-
-# workflows:
-#   build:
-#     jobs:
-#       - setup:
-#           context: molecula
-#           filters:
-#             tags:
-#               only: /^v.*/
-#       - linter:
-#           context: molecula
-#           requires:
-#             - setup
-#       - go-mod-tidy:
-#           context: molecula
-#           requires:
-#             - setup
-#       - check-changelog-label:
-#           context: molecula
-#           requires:
-#             - setup
-#       - test-build-arm:
-#           context: molecula
-#           requires:
-#             - setup
-#       - test:
-#           name: test-golang-<< matrix.golang_version >>
-#           resource_class: large
-#           context: molecula
-#           requires:
-#             - setup
-#           matrix:
-#             parameters:
-#               golang_version: ["1.15.8", "1.16.10"]
-#       - test:
-#           name: << matrix.test_make_target >>
-#           resource_class: xlarge
-#           context: molecula
-#           requires:
-#             - setup
-#           matrix:
-#             parameters:
-#               test_make_target: ["test-race"]
-#       - test:
-#           name: test-shardwidth-22
-#           context: molecula
-#           shard_width: "22"
-#           resource_class: large
-#           requires:
-#             - setup
-#       - test-external-lookup:
-#           context: molecula
-#           requires:
-#             - setup
-#       - cluster-tests:
-#           context: molecula
-#           requires:
-#             - setup
-#       - docker-build:
-#           context: molecula
-#           requires:
-#             - setup
-#       - release:
-#           context: molecula
-#           requires:
-#             - setup
-#           filters:
-#             tags:
-#               only: /^v.*/
-#       - publish_release:
-#           context: molecula
-#           requires:
-#             - release
-#           filters:
-#             tags:
-#               only: /^v.*/
-#             branches:
-#               ignore: /.*/
-#       - dockerhub-upload-unstable:
-#           context: molecula
-#           requires:
-#             - setup
-#           filters:
-#             branches:
-#               only: master
-#       - dockerhub-upload-stable:
-#           context: molecula
-#           requires:
-#             - setup
-#           filters:
-#             tags:
-#               only: /^v.*/
-#             branches:
-#               ignore: /.*/

.gitignore

@@ -20,4 +20,5 @@ launch.json
 __pycache__/
 report.xml
 outputs.json
-builds/
+builds/
+*.tfstate.backup

.gitlab/.gitlab-ci.yml

@@ -4,7 +4,7 @@ include:
   - template: Security/Dependency-Scanning.gitlab-ci.yml
 
 variables:
-  GOVERSION: "1.16.13"
+  GOVERSION: "1.17.7"
 
 stages:
   - lint
@@ -12,7 +12,9 @@ stages:
   - build
   - integration
   - gauntlet
+  - performance
   - post build
+  - nonblocking
 
 smoke build:
   image: golang:$GOVERSION
@@ -34,6 +36,15 @@ golangci-lint:
     - echo "Checking for issues in new code"
     - golangci-lint run
 
+go mod tidy:
+  stage: lint
+  image: golang:$GOVERSION
+  rules:
+    - if: '$CI_COMMIT_TAG == null && ($CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web")'
+  script:
+    - go mod tidy
+    - git diff --exit-code -- go.mod go.sum
+
 build lattice:
   stage: test
   image: node:14
@@ -83,11 +94,12 @@ run go tests:
     - aws
 
 run go tests race:
-  stage: test
+  stage: nonblocking # don't let this job block any other jobs because it takes much longer than the other tests.
   image: golang:$GOVERSION
   rules:
     - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
   retry: 1
+  needs: [] # don't wait to start running this.
   script:
     - echo "Running featurebase race tests..."
     - go test -race -v -timeout=90m ./...
@@ -100,7 +112,7 @@ run go tests shardwidth22:
   rules:
     - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
   script:
-    - echo "Running featurebase race tests..."
+    - echo "Running featurebase shardwidth22 tests..."
     - go test -timeout=30m -tags=shardwidth22 ./...
   tags:
     - aws
@@ -287,6 +299,7 @@ clustertests:
   stage: integration
   tags:
     - shell
+  retry: 1
   rules:
     - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
   script:
@@ -301,6 +314,7 @@ authclustertests:
   variables:
     PROJECT: authclustertests_${CI_CONCURRENT_ID}
   stage: integration
+  retry: 1
   tags:
     - shell
   rules:
@@ -337,7 +351,6 @@ smoke test:
     AWS_ACCESS_KEY_ID: $AWS_FBCI_ACCESS_KEY_ID
     AWS_SECRET_ACCESS_KEY: $AWS_FBCI_SECRET_ACCESS_KEY
     TF_VAR_cluster_prefix: ""
-    TF_VAR_branch: ""
   tags:
     - aws
     - docker
@@ -369,8 +382,6 @@ smoke test:
     - export PATH=$PATH:/usr/local/go/bin
     - TF_VAR_cluster_prefix="smoke-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)"
     - echo "Cluster Prefix --> $TF_VAR_cluster_prefix"
-    - TF_VAR_branch=$CI_COMMIT_BRANCH
-    - echo "Branch --> $TF_VAR_branch"
   script:
     - ./qa/scripts/setupSmokeTest.sh
     - ./qa/scripts/testSmokeTest.sh
@@ -398,7 +409,6 @@ gauntlet:
     AWS_SECRET_ACCESS_KEY: $AWS_FBCI_SECRET_ACCESS_KEY
     ASG_NAME: "gitlab-runners"
     TF_VAR_cluster_prefix: ""
-    TF_VAR_branch: ""
   tags:
     - aws
     - docker
@@ -433,8 +443,6 @@ gauntlet:
     - export PATH=$PATH:/usr/local/go/bin
     - TF_VAR_cluster_prefix="gauntlet-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)"
     - echo "Cluster Prefix --> $TF_VAR_cluster_prefix"
-    - TF_VAR_branch=$CI_COMMIT_BRANCH
-    - echo "Branch --> $TF_VAR_branch"
     - export INSTANCE_ID=$(curl --silent --fail "http://169.254.169.254/latest/meta-data/instance-id" | tee instance_id)
     - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --protected-from-scale-in --profile $INFRA_PROFILE
   script:
@@ -455,7 +463,7 @@ s3 dump:
   tags:
     - shell
   rules:
-    - if: '$CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "schedule" || $CI_PIPELINE_SOURCE == "web"'
+    - if: '$CI_COMMIT_TAG == null && ($CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "web")'
   script:
     - aws configure set aws_access_key_id $AWS_FBCI_ACCESS_KEY_ID
     - aws configure set aws_secret_access_key $AWS_FBCI_SECRET_ACCESS_KEY
@@ -482,3 +490,50 @@ s3 dump:
     - job: build for darwin arm64
     - job: build for linux amd64
     - job: build for linux arm64
+
+perf_able:
+  stage: performance
+  rules:
+    - if: '$CI_COMMIT_BRANCH == $CI_DEFAULT_BRANCH && $CI_PIPELINE_SOURCE == "push"'
+  trigger:
+    include: .gitlab/.perf-able-gitlab-ci.yml
+  variables:
+    PARENT_PIPELINE_ID: $CI_PIPELINE_ID
+
+s3 dump tag:
+  stage: post build
+  variables:
+    PROFILE: "service-fb-ci"
+    AWS_SSH_PRIVATE_KEY: $AWS_FBCI_SSH_KEY
+    AWS_ACCESS_KEY_ID: $AWS_FBCI_ACCESS_KEY_ID
+    AWS_SECRET_ACCESS_KEY: $AWS_FBCI_SECRET_ACCESS_KEY
+    LOCATION: molecula-artifact-storage/featurebase/_tags
+  tags:
+    - shell
+  rules:
+    - if: '$CI_COMMIT_TAG != null && ($CI_PIPELINE_SOURCE == "push" || $CI_PIPELINE_SOURCE == "web")'
+  script:
+    - aws configure set aws_access_key_id $AWS_FBCI_ACCESS_KEY_ID
+    - aws configure set aws_secret_access_key $AWS_FBCI_SECRET_ACCESS_KEY
+    - aws configure set region "us-east-2"
+    - aws configure set aws_profile $PROFILE
+    - |
+      for goos in "darwin" "linux"; do
+        for goarch in "amd64" "arm64"; do
+          dir=featurebase-${CI_COMMIT_TAG}-${goos}-${goarch}
+          echo "Directory ${dir}"
+          mkdir $dir
+          mv featurebase_${goos}_${goarch} ${dir}/featurebase
+          mv roaring-migrate_${goos}_${goarch} ${dir}/roaring-migrate
+          cp NOTICE install/featurebase.conf install/featurebase.*.service ${dir}/
+          tar cvzf ${dir}.tar.gz ${dir}
+          aws s3 cp ${dir} s3://${LOCATION}/${CI_COMMIT_TAG}/${dir}/ --recursive
+          aws s3 cp ${dir}.tar.gz s3://${LOCATION}/${CI_COMMIT_TAG}/
+        done
+      done
+
+  needs:
+    - job: build for darwin amd64
+    - job: build for darwin arm64
+    - job: build for linux amd64
+    - job: build for linux arm64

.gitlab/.perf-able-gitlab-ci.yml

@@ -0,0 +1,61 @@
+stages:
+  - performance
+
+perf_able:
+  stage: performance
+  timeout: 2h
+  image: registry.gitlab.com/gitlab-org/cloud-deploy/aws-base:latest
+  variables:
+    PROFILE: "service-terraform"
+    INFRA_PROFILE: "service-gitlab"
+    AWS_SSH_PRIVATE_KEY: $AWS_FBCI_SSH_KEY
+    AWS_ACCESS_KEY_ID: $AWS_FBCI_ACCESS_KEY_ID
+    AWS_SECRET_ACCESS_KEY: $AWS_FBCI_SECRET_ACCESS_KEY
+    ASG_NAME: "gitlab-runners"
+    TF_VAR_cluster_prefix: ""
+  tags:
+    - aws
+    - docker
+    - fbsmoke
+  before_script:
+    - apt-get update && apt-get install -y gnupg software-properties-common curl git
+    - curl -fsSL https://apt.releases.hashicorp.com/gpg | apt-key add -
+    - apt-add-repository "deb [arch=amd64] https://apt.releases.hashicorp.com $(lsb_release -cs) main"
+    - apt-get update && apt-get install terraform
+    - aws configure set aws_access_key_id $AWS_FBCI_ACCESS_KEY_ID --profile $PROFILE
+    - aws configure set aws_secret_access_key $AWS_FBCI_SECRET_ACCESS_KEY --profile $PROFILE
+    - aws configure set region "us-east-2" --profile $PROFILE
+    - aws configure set aws_profile $PROFILE
+    - aws configure set aws_access_key_id $AWS_INFRA_ACCESS_KEY_ID --profile $INFRA_PROFILE
+    - aws configure set aws_secret_access_key $AWS_INFRA_SECRET_ACCESS_KEY --profile $INFRA_PROFILE
+    - aws configure set region "us-east-2" --profile $INFRA_PROFILE
+    - echo $AWS_FBCI_SSH_KEY > gitlab-featurebase-ci.pem
+    - chmod 400 gitlab-featurebase-ci.pem
+    - 'which ssh-agent || ( apt-get update -y && apt-get install openssh-client -y )'
+    - eval $(ssh-agent -s)
+    - mkdir -p ~/.ssh
+    - echo $AWS_FBCI_SSH_KEY > /root/.ssh/gitlab-featurebase-ci.pem
+    - chmod 400 /root/.ssh/gitlab-featurebase-ci.pem
+    - echo "$AWS_FBCI_SSH_KEY" | ssh-add -
+    - chmod 700 /root/.ssh
+    - '[[ -f /.dockerenv ]] && echo -e "Host *\n\tStrictHostKeyChecking no\n\n" > ~/.ssh/config'
+    - apt update && apt -y install jq wget
+    - wget -q https://go.dev/dl/go1.17.5.linux-amd64.tar.gz
+    - tar -C /usr/local -xzf go1.17.5.linux-amd64.tar.gz
+    - export PATH=$PATH:/usr/local/go/bin
+    - TF_VAR_cluster_prefix="able-$(openssl rand -base64 12 | tr -d /=+ | cut -c -16)"
+    - echo "Cluster Prefix --> $TF_VAR_cluster_prefix"
+    - export INSTANCE_ID=$(curl --silent --fail "http://169.254.169.254/latest/meta-data/instance-id" | tee instance_id)
+    - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --protected-from-scale-in --profile $INFRA_PROFILE
+  script:
+    - ./qa/scripts/perf/able/ableSetup.sh
+    - ./qa/scripts/perf/able/ableTest.sh
+  after_script:
+    - ./qa/scripts/perf/able/ableTeardown.sh || true
+    - export INSTANCE_ID=$(cat instance_id)
+    - aws autoscaling set-instance-protection --instance-ids "$INSTANCE_ID" --auto-scaling-group-name $ASG_NAME --no-protected-from-scale-in --profile $INFRA_PROFILE
+  needs:
+    - pipeline: $PARENT_PIPELINE_ID
+      job: build for linux arm64
+    - pipeline: $PARENT_PIPELINE_ID
+      job: build for linux amd64

Dockerfile-clustertests

@@ -1,7 +1,7 @@
 # This Dockerfile is used for cluster testing - it produces a much larger image
 # and includes all of Go as well as some utilities.
 
-FROM golang:1.16
+FROM golang:1.17
 
 LABEL maintainer "dev@pilosa.com"
 

Dockerfile-clustertests-client

@@ -1,7 +1,7 @@
 # This Dockerfile is used for cluster testing - it produces a much larger image
 # and includes all of Go as well as some utilities.
 
-FROM golang:1.16
+FROM golang:1.17
 
 LABEL maintainer "dev@pilosa.com"
 

Makefile

@@ -13,7 +13,7 @@ BUILD_TIME := $(shell date -u +%FT%T%z)
 SHARD_WIDTH = 20
 COMMIT := $(shell git describe --exact-match >/dev/null 2>&1 || git rev-parse --short HEAD)
 LDFLAGS="-X github.com/molecula/featurebase/v3.Version=$(VERSION) -X github.com/molecula/featurebase/v3.BuildTime=$(BUILD_TIME) -X github.com/molecula/featurebase/v3.Variant=$(VARIANT) -X github.com/molecula/featurebase/v3.Commit=$(COMMIT) -X github.com/molecula/featurebase/v3.TrialDeadline=$(TRIAL_DEADLINE)"
-GO_VERSION=1.16.10
+GO_VERSION=1.17.7
 DOCKER_BUILD= # set to 1 to use `docker-build` instead of `build` when creating a release
 BUILD_TAGS += shardwidth$(SHARD_WIDTH)
 TEST_TAGS = roaringparanoia

api.go

@@ -50,9 +50,6 @@ type API struct {
 	importWorkerPoolSize int
 	importWork           chan importJob
 
-	usageCache      *usageCache
-	schemaDetailsOn bool
-
 	Serializer Serializer
 }
 
@@ -73,14 +70,6 @@ func OptAPIServer(s *Server) apiOption {
 	}
 }
 
-// Used to configure API option: schemaDetailsOn
-func OptAPISchemaDetailsOn(isOn bool) apiOption {
-	return func(a *API) error {
-		a.schemaDetailsOn = isOn
-		return nil
-	}
-}
-
 func OptAPIImportWorkerPoolSize(size int) apiOption {
 	return func(a *API) error {
 		a.importWorkerPoolSize = size
@@ -339,11 +328,17 @@ func (api *API) CreateField(ctx context.Context, indexName string, fieldName str
 	}
 
 	// Create field.
-	field, err := index.CreateFieldAndBroadcast(cfm)
+	field, err := index.CreateField(fieldName, opts...)
 	if err != nil {
 		return nil, errors.Wrap(err, "creating field")
 	}
 
+	// Send the create field message to all nodes. We do this *outside* the
+	// CreateField logic so we're not blocking on it.
+	if err := api.holder.sendOrSpool(cfm); err != nil {
+		return nil, errors.Wrap(err, "sending CreateField message")
+	}
+
 	api.holder.Stats.CountWithCustomTags(MetricCreateField, 1, 1.0, []string{fmt.Sprintf("index:%s", indexName)})
 	return field, nil
 }
@@ -938,256 +933,6 @@ func (api *API) PrimaryNode() *topology.Node {
 	return snap.PrimaryFieldTranslationNode()
 }
 
-// Cache of disk usage statistics
-type usageCache struct {
-	data             map[string]NodeUsage
-	refreshInterval  time.Duration
-	lastUpdated      time.Time
-	resetTrigger     chan bool
-	lastCalcDuration time.Duration
-	waitMultiplier   float64
-	disable          bool
-
-	muCalculate sync.Mutex
-	muAssign    sync.Mutex
-}
-
-var usageCacheMinDuration = 5 * time.Second // If usage takes less than this duration to calculate, don't use the cache.
-var usageCacheMinInterval = time.Hour       // Refresh interval is forced to be >= this duration.
-var usageCacheInitialInterval = time.Hour   // Refresh interval starts with this duration.
-
-// NodeUsage represents all usage measurements for one node.
-type NodeUsage struct {
-	Disk        DiskUsage   `json:"diskUsage"`
-	Memory      MemoryUsage `json:"memoryUsage"`
-	LastUpdated time.Time   `json:"lastUpdated"`
-}
-
-// DiskUsage represents the storage space used on disk by one node.
-type DiskUsage struct {
-	Capacity   uint64                `json:"capacity,omitempty"`
-	TotalUse   uint64                `json:"totalInUse"`
-	IndexUsage map[string]IndexUsage `json:"indexes"`
-}
-
-// IndexUsage represents the storage space used on disk by one index, on one node.
-type IndexUsage struct {
-	Total          uint64                `json:"total"`
-	IndexKeys      uint64                `json:"indexKeys"`
-	FieldKeysTotal uint64                `json:"fieldKeysTotal"`
-	Fragments      uint64                `json:"fragments"`
-	Metadata       uint64                `json:"metadata"`
-	Fields         map[string]FieldUsage `json:"fields"`
-}
-
-// FieldUsage represents the storage space used on disk by one field, on one node
-type FieldUsage struct {
-	Total     uint64 `json:"total"`
-	Fragments uint64 `json:"fragments"`
-	Keys      uint64 `json:"keys"`
-	Metadata  uint64 `json:"metadata"`
-}
-
-// MemoryUsage represents the memory used by one node.
-type MemoryUsage struct {
-	Capacity uint64 `json:"capacity"`
-	TotalUse uint64 `json:"totalInUse"`
-}
-
-// Returns disk usage from cache if cache is large. It will recalculate on the spot if the last cacluation was under 5 seconds.
-func (api *API) Usage(ctx context.Context, remote bool) (map[string]NodeUsage, error) {
-	span, _ := tracing.StartSpanFromContext(ctx, "API.Usage")
-	defer span.Finish()
-
-	if api.usageCache.disable {
-		resp := make(map[string]NodeUsage)
-		return resp, nil
-	}
-
-	api.usageCache.muAssign.Lock()
-	lastCalc := api.usageCache.lastCalcDuration
-	api.usageCache.muAssign.Unlock()
-	if lastCalc < usageCacheMinDuration {
-		err := api.ResetUsageCache()
-		if err != nil {
-			api.server.logger.Infof("could not reset usageCache: %s", err)
-		}
-	}
-
-	api.usageCache.muAssign.Lock()
-	lastUpdated := api.usageCache.lastUpdated
-	api.usageCache.muAssign.Unlock()
-	if lastUpdated == (time.Time{}) {
-		api.calculateUsage()
-	}
-
-	if !remote {
-		api.requestUsageOfNodes()
-	}
-
-	return api.usageCache.data, nil
-}
-
-// Makes a ui/usage request for each node in cluster to calculates its usage and adds it to the cache
-func (api *API) requestUsageOfNodes() {
-	nodes := api.cluster.Nodes()
-	for _, node := range nodes {
-		if node.ID == api.server.nodeID {
-			continue
-		}
-
-		nodeUsage, err := api.server.defaultClient.GetNodeUsage(context.Background(), &node.URI)
-		if err != nil {
-			api.server.logger.Infof("couldn't collect disk usage from %s: %s", node.URI, err)
-		}
-
-		api.usageCache.muAssign.Lock()
-		api.usageCache.data[node.ID] = nodeUsage[node.ID]
-		api.usageCache.muAssign.Unlock()
-	}
-}
-
-// Calculates disk usage from scratch if cache has expired for each index and stores the results in the usage cache
-func (api *API) calculateUsage() {
-	api.usageCache.muCalculate.Lock()
-	defer api.usageCache.muCalculate.Unlock()
-	api.server.wg.Add(1)
-	defer api.server.wg.Done()
-
-	api.usageCache.muAssign.Lock()
-	lastUpdated := api.usageCache.lastUpdated
-	api.usageCache.muAssign.Unlock()
-
-	if time.Since(lastUpdated) <= api.usageCache.refreshInterval {
-		return
-	}
-	indexDetails, nodeMetadataBytes, err := api.holder.Txf().IndexUsageDetails(api.isClosing)
-	if err != nil {
-		api.server.logger.Infof("couldn't get index usage details: %s", err)
-	}
-	if api.isClosing() {
-		return
-	}
-
-	totalSize := nodeMetadataBytes
-	for _, s := range indexDetails {
-		totalSize += s.Total
-	}
-
-	// NOTE: these errors are ignored in api.Info(), but checked here
-	si := api.server.systemInfo
-	diskCapacity, err := si.DiskCapacity(api.holder.path)
-	if err != nil {
-		api.server.logger.Infof("couldn't read disk capacity: %s", err)
-	}
-
-	memoryCapacity, err := si.MemTotal()
-	if err != nil {
-		api.server.logger.Infof("couldn't read memory capacity: %s", err)
-	}
-	memoryUse, err := si.MemUsed()
-	if err != nil {
-		api.server.logger.Infof("couldn't read memory usage: %s", err)
-	}
-
-	lastUpdated = time.Now()
-	// Insert into result.
-	nodeUsage := NodeUsage{
-		Disk: DiskUsage{
-			Capacity:   diskCapacity,
-			TotalUse:   totalSize,
-			IndexUsage: indexDetails,
-		},
-		Memory: MemoryUsage{
-			Capacity: memoryCapacity,
-			TotalUse: memoryUse,
-		},
-		LastUpdated: lastUpdated,
-	}
-	api.usageCache.muAssign.Lock()
-	api.usageCache.data = make(map[string]NodeUsage)
-	api.usageCache.data[api.server.nodeID] = nodeUsage
-	api.usageCache.lastUpdated = lastUpdated
-	api.usageCache.muAssign.Unlock()
-}
-
-// Periodically calculates disk/memory usage in terms of the duty cycle. The duty cycle represents the percentage of
-// time that is spent recalculating this cache. It is specified relatively, rather than by a set interval, because
-// scans can take an unpredictably long time.
-func (api *API) RefreshUsageCache(dutyCycle float64) {
-
-	if dutyCycle == 0 {
-		api.server.logger.Warnf("usage-duty-cycle set to 0, usage cache and /ui/usage endpoint are disabled")
-		api.usageCache = &usageCache{
-			disable: true,
-		}
-		return
-	}
-
-	trigger := make(chan bool)
-	defer close(trigger)
-
-	multiplier := 100/dutyCycle - 1
-
-	api.usageCache = &usageCache{
-		data:             make(map[string]NodeUsage),
-		refreshInterval:  usageCacheInitialInterval,
-		resetTrigger:     trigger,
-		lastCalcDuration: 0,
-		waitMultiplier:   multiplier,
-	}
-	api.server.logger.Infof("monitoring resource usage with duty cycle %v%%\n", dutyCycle)
-	for {
-		start := time.Now()
-		api.calculateUsage()
-		api.setRefreshInterval(time.Since(start))
-		api.server.logger.Infof("updated resource usage cache at %v, took %v, next update in %v\n", api.usageCache.lastUpdated.Format(time.RFC3339), api.usageCache.lastCalcDuration.Truncate(time.Millisecond), api.usageCache.refreshInterval.Truncate(100*time.Millisecond))
-		select {
-		case <-trigger:
-			continue
-		case <-api.server.closing:
-			return
-		case <-time.After(api.usageCache.refreshInterval):
-			continue
-		}
-	}
-}
-
-// Refresh interval set in relation to how long the last calculation took.
-func (api *API) setRefreshInterval(dur time.Duration) {
-	refresh := time.Duration(float64(dur) * api.usageCache.waitMultiplier)
-	if refresh < usageCacheMinInterval {
-		refresh = usageCacheMinInterval
-	}
-	api.usageCache.muAssign.Lock()
-	api.usageCache.refreshInterval = refresh
-	api.usageCache.lastCalcDuration = dur
-	api.usageCache.muAssign.Unlock()
-}
-
-// Resets the lastUpdated time and awakens RefreshUsageCache()
-func (api *API) ResetUsageCache() error {
-	if api.usageCache != nil {
-		api.usageCache.muAssign.Lock()
-		api.usageCache.lastUpdated = time.Time{}
-		api.usageCache.muAssign.Unlock()
-	} else {
-		return errors.New("invalidating cache: cache not initialized")
-	}
-	api.usageCache.resetTrigger <- true
-	return nil
-}
-
-// isClosing returns true if the server is shutting down.
-func (api *API) isClosing() bool {
-	select {
-	case <-api.server.closing:
-		return true
-	default:
-		return false
-	}
-}
-
 // RecalculateCaches forces all TopN caches to be updated.
 // This is done internally within a TopN query, but a user may want to do it ahead of time?
 func (api *API) RecalculateCaches(ctx context.Context) error {
@@ -1272,38 +1017,6 @@ func (api *API) Schema(ctx context.Context, withViews bool) ([]*IndexInfo, error
 	return api.holder.limitedSchema()
 }
 
-// SchemaDetails returns information about each index in Pilosa including which
-// fields they contain. Additional field information such as cardinality unless
-// turned off via the schemaDetailsOn cli option.
-func (api *API) SchemaDetails(ctx context.Context) ([]*IndexInfo, error) {
-	span, _ := tracing.StartSpanFromContext(ctx, "API.Schema")
-	defer span.Finish()
-	schema, err := api.holder.Schema()
-	if err != nil {
-		return nil, errors.Wrap(err, "getting schema")
-	}
-	if !api.schemaDetailsOn {
-		return schema, nil
-	}
-	for _, index := range schema {
-		for _, field := range index.Fields {
-			q := fmt.Sprintf("Count(Distinct(field=%s))", field.Name)
-			req := QueryRequest{Index: index.Name, Query: q}
-			resp, err := api.query(ctx, &req)
-			if err != nil {
-				return schema, errors.Wrapf(err, "querying cardinality (%s/%s)", index.Name, field.Name)
-			}
-			if len(resp.Results) == 0 {
-				continue
-			}
-			if card, ok := resp.Results[0].(uint64); ok {
-				field.Cardinality = &card
-			}
-		}
-	}
-	return schema, nil
-}
-
 // ApplySchema takes the given schema and applies it across the
 // cluster (if remote is false), or just to this node (if remote is
 // true). This is designed for the use case of replicating a schema
@@ -1326,6 +1039,164 @@ func (api *API) ApplySchema(ctx context.Context, s *Schema, remote bool) error {
 	return nil
 }
 
+// applyOneIngestSchema applies a single ingestSpec, which specifies operations on
+// a single index and possibly fields. If it is successful, it returns the name
+// of the index and an empty slice (if it created the index), or the name of the
+// index and a slice of the fields within that index that it created. If it
+// is unsuccessful, it tries to delete whatever it created.
+//
+// The intended idiom is that if the returned list of fields isn't empty, the index
+// already existed and only those fields need to be cleaned up in the event of
+// a later error, but if the list of fields is empty, the entire index was new,
+// and should be cleaned up, in which case there's no need to track or delete
+// the specific fields separately.
+func (api *API) ApplyOneIngestSchema(ctx context.Context, schema *ingestSpec) (index *Index, returnedFields []string, err error) {
+	if api.PrimaryNode().ID != api.NodeID() {
+		return nil, nil, RedirectError{
+			HostPort: api.PrimaryNode().URI.Normalize(),
+			error:    "request made to non-primary node",
+		}
+	}
+
+	// create index
+	indexName := schema.IndexName
+	var createdFields []string
+	var useKeys bool
+	switch schema.PrimaryKeyType {
+	case "string":
+		useKeys = true
+	case "uint":
+		useKeys = false
+	default:
+		return nil, nil, fmt.Errorf("invalid primary key type %q", schema.PrimaryKeyType)
+	}
+	opts := IndexOptions{
+		Keys:           useKeys,
+		TrackExistence: true,
+	}
+	createdIndex := false
+
+	// We check this up here because, if there's at least one field but we don't know what to do with
+	// it, we will necessarily fail, which means we'd delete the index anyway, so there's no point in
+	// trying to create it. We don't care about this if there's no fields specified.
+	if len(schema.Fields) > 0 {
+		switch schema.FieldAction {
+		case "create", "ensure", "require":
+			// do nothing
+		case "":
+			schema.FieldAction = schema.IndexAction
+		default:
+			return nil, nil, fmt.Errorf("invalid field-action %q, expecting create/ensure/require", schema.FieldAction)
+		}
+	}
+
+	switch schema.IndexAction {
+	case "ensure", "require":
+		index, err = api.Index(ctx, indexName)
+		if err != nil {
+			if _, ok := err.(NotFoundError); !ok {
+				return nil, nil, fmt.Errorf("checking for existing index %q: %w", indexName, err)
+			} else {
+				err = nil
+			}
+		}
+		if index != nil {
+			existingOpts := index.Options()
+			if existingOpts != opts {
+				return nil, nil, fmt.Errorf("index %q options mismatch: schema %#v, existing %#v", indexName, opts, existingOpts)
+			}
+			break
+		}
+		if schema.IndexAction == "require" {
+			return nil, nil, fmt.Errorf("index %q does not exist", indexName)
+		}
+		fallthrough
+	case "create":
+		index, err = api.CreateIndex(ctx, indexName, opts)
+		if err != nil {
+			return nil, nil, err
+		}
+		createdIndex = true
+	default:
+		return nil, nil, fmt.Errorf("invalid index-action %q, need create/ensure/require", schema.IndexAction)
+	}
+
+	// Now we might have an index, so we need our cleanup code.
+	defer func() {
+		if err == nil {
+			return
+		}
+		if createdIndex {
+			err := api.DeleteIndex(ctx, indexName)
+			if err != nil {
+
+				api.server.logger.Printf("trying to undo failed index %q creation: %v", indexName, err)
+			}
+			return
+		}
+		for _, field := range createdFields {
+			err := api.DeleteField(ctx, indexName, field)
+			if err != nil {
+				api.server.logger.Printf("trying to undo failed field %q creation in index %q: %v", field, indexName, err)
+			}
+		}
+	}()
+
+	// create all the fields specified in the index
+	for _, fSpec := range schema.Fields {
+		fieldName := fSpec.FieldName
+		opt := fieldSpecToFieldOption(fSpec)
+		err = opt.validate()
+		if err != nil {
+			return nil, nil, err
+		}
+		switch schema.FieldAction {
+		case "ensure", "require":
+			field, schemaErr := api.Field(ctx, indexName, fieldName)
+			if schemaErr != nil {
+				// NotFoundError is fine
+				if _, ok := schemaErr.(NotFoundError); !ok {
+					return nil, nil, fmt.Errorf("checking for existing field %q in %q: %w", fieldName, indexName, err)
+				}
+			}
+			if field != nil {
+				existing := field.Options()
+				if opt.Type != existing.Type {
+					return nil, nil, fmt.Errorf("existing field %q is %q, not %q", fieldName, existing.Type, opt.Type)
+				}
+				if ((opt.Keys != nil) && *opt.Keys) != existing.Keys {
+					if existing.Keys {
+						return nil, nil, fmt.Errorf("existing field %q in %q uses keys", fieldName, indexName)
+					} else {
+						return nil, nil, fmt.Errorf("existing field %q in %q doesn't use keys", fieldName, indexName)
+					}
+				}
+				// TODO: verify compatibility of other field opts, this is sorta hard
+				break
+			}
+			if schema.FieldAction == "require" {
+				return nil, nil, fmt.Errorf("field %q does not exist in %q", fieldName, indexName)
+			}
+			fallthrough
+		case "create":
+			fos := fieldOptionsToFunctionalOpts(opt)
+			_, err = api.CreateField(ctx, indexName, fieldName, fos...)
+			if err != nil {
+				return nil, nil, fmt.Errorf("creating field %q in %q: %v", fieldName, indexName, err)
+			}
+			createdFields = append(createdFields, fieldName)
+		}
+	}
+
+	// we don't report the fields back, so we can distinguish "created index"
+	// from "created fields within index"
+	if createdIndex {
+		createdFields = nil
+	}
+
+	return index, createdFields, nil
+}
+
 // Views returns the views in the given field.
 func (api *API) Views(ctx context.Context, indexName string, fieldName string) ([]*view, error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "API.Views")
@@ -1905,6 +1776,13 @@ func (api *API) IngestOperations(ctx context.Context, qcx *Qcx, indexName string
 	span, _ := tracing.StartSpanFromContext(ctx, "API.IngestOperations")
 	defer span.Finish()
 
+	if api.PrimaryNode().ID != api.NodeID() {
+		return RedirectError{
+			HostPort: api.PrimaryNode().URI.Normalize(),
+			error:    "request made to non-primary node",
+		}
+	}
+
 	if err := api.validate(apiIngestOperations); err != nil {
 		return errors.Wrap(err, "validating api method")
 	}
@@ -3252,24 +3130,24 @@ var methodsResizing = map[apiMethod]struct{}{
 	apiSchema:             {},
 }
 
-var methodsDegraded = map[apiMethod]struct{}{
-	apiExportCSV:         {},
-	apiFragmentBlockData: {},
-	apiFragmentBlocks:    {},
-	apiField:             {},
-	apiIndex:             {},
-	apiQuery:             {},
-	apiRecalculateCaches: {},
-	apiRemoveNode:        {},
-	apiShardNodes:        {},
-	apiSchema:            {},
-	apiViews:             {},
-	apiStartTransaction:  {},
-	apiFinishTransaction: {},
-	apiTransactions:      {},
-	apiGetTransaction:    {},
-	apiActiveQueries:     {},
-}
+// var methodsDegraded = map[apiMethod]struct{}{
+// 	apiExportCSV:         {},
+// 	apiFragmentBlockData: {},
+// 	apiFragmentBlocks:    {},
+// 	apiField:             {},
+// 	apiIndex:             {},
+// 	apiQuery:             {},
+// 	apiRecalculateCaches: {},
+// 	apiRemoveNode:        {},
+// 	apiShardNodes:        {},
+// 	apiSchema:            {},
+// 	apiViews:             {},
+// 	apiStartTransaction:  {},
+// 	apiFinishTransaction: {},
+// 	apiTransactions:      {},
+// 	apiGetTransaction:    {},
+// 	apiActiveQueries:     {},
+// }
 
 var methodsNormal = map[apiMethod]struct{}{
 	apiCreateField:          {},

api_test.go

@@ -29,6 +29,8 @@ import (
 	"github.com/molecula/featurebase/v3/shardwidth"
 	"github.com/molecula/featurebase/v3/test"
 	. "github.com/molecula/featurebase/v3/vprint" // nolint:staticcheck
+
+	"golang.org/x/sync/errgroup"
 )
 
 func TestAPI_Import(t *testing.T) {
@@ -553,7 +555,7 @@ func TestAPI_Ingest(t *testing.T) {
 	if err != nil {
 		t.Fatalf("creating field: %v", err)
 	}
-	_, err = coord.API.CreateField(ctx, index, timeField, pilosa.OptFieldTypeTime("YMD"))
+	_, err = coord.API.CreateField(ctx, index, timeField, pilosa.OptFieldTypeTime("YMD", "0"))
 	if err != nil {
 		t.Fatalf("creating field: %v", err)
 	}
@@ -676,7 +678,7 @@ func BenchmarkIngest(b *testing.B) {
 	if err != nil {
 		b.Fatalf("creating field: %v", err)
 	}
-	_, err = coord.API.CreateField(ctx, index, tqField, pilosa.OptFieldTypeTime("YMDH"))
+	_, err = coord.API.CreateField(ctx, index, tqField, pilosa.OptFieldTypeTime("YMDH", "0"))
 	if err != nil {
 		b.Fatalf("creating field: %v", err)
 	}
@@ -956,29 +958,6 @@ func TestAPI_IDAlloc(t *testing.T) {
 	})
 }
 
-func TestAPI_SchemaDetailsOff(t *testing.T) {
-	cluster := test.MustRunCluster(t, 2)
-	defer cluster.Close()
-	cmd := cluster.GetNode(0)
-	err := cmd.API.SetAPIOptions(pilosa.OptAPISchemaDetailsOn(false))
-	if err != nil {
-		t.Fatalf("could not toggle schema details to off: %v", err)
-	}
-	schema, err := cmd.API.SchemaDetails(context.Background())
-	if err != nil {
-		t.Fatalf("getting schema: %v", err)
-	}
-
-	for _, i := range schema {
-		for _, f := range i.Fields {
-			if f.Cardinality != nil {
-				t.Fatalf("expected nil cardinality, got: %v", *f.Cardinality)
-			}
-		}
-	}
-
-}
-
 type mutexCheckIndex struct {
 	index     *pilosa.Index
 	indexName string
@@ -1426,6 +1405,42 @@ func TestVariousApiTranslateCalls(t *testing.T) {
 	}
 }
 
+func TestAPI_CreateField(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	c := test.MustRunCluster(t, 3)
+	defer c.Close()
+
+	nodes := make([]*test.Command, 3)
+	for i := range nodes {
+		nodes[i] = c.GetNode(i)
+	}
+
+	if _, err := nodes[0].API.CreateIndex(ctx, "i", pilosa.IndexOptions{}); err != nil {
+		t.Fatal(err)
+	}
+	eg, ctx := errgroup.WithContext(context.Background())
+	for _, n := range nodes {
+		node := n
+		eg.Go(func() error {
+			for i := 0; i < 10; i++ {
+				_, err := node.API.CreateField(ctx, "i", fmt.Sprintf("f%d", i))
+				if err != nil && !errors.Is(err, pilosa.ErrFieldExists) {
+					return err
+				}
+			}
+			return nil
+		})
+	}
+	err := eg.Wait()
+	if err != nil {
+		if errors.Is(err, pilosa.ErrFieldExists) {
+			t.Fatalf("conflict error: %v", err)
+		}
+		t.Fatalf("unexpected error: %T %v", err, err)
+	}
+}
+
 func TestAPI_RBFDebugInfo(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()

apimethod_string.go

@@ -43,11 +43,15 @@ func _() {
 	_ = x[apiIDReserve-32]
 	_ = x[apiIDCommit-33]
 	_ = x[apiIDReset-34]
+	_ = x[apiPartitionNodes-35]
+	_ = x[apiIngestOperations-36]
+	_ = x[apiIngestNodeOperations-37]
+	_ = x[apiMutexCheck-38]
 }
 
-const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDReset"
+const _apiMethod_name = "apiClusterMessageapiCreateFieldapiCreateIndexapiDeleteFieldapiDeleteAvailableShardapiDeleteIndexapiDeleteViewapiExportCSVapiFragmentBlockDataapiFragmentBlocksapiFragmentDataapiTranslateDataapiFieldTranslateDataapiFieldapiImportapiImportValueapiIndexapiQueryapiRecalculateCachesapiRemoveNodeapiResizeAbortapiSchemaapiShardNodesapiStateapiViewsapiApplySchemaapiStartTransactionapiFinishTransactionapiTransactionsapiGetTransactionapiActiveQueriesapiPastQueriesapiIDReserveapiIDCommitapiIDResetapiPartitionNodesapiIngestOperationsapiIngestNodeOperationsapiMutexCheck"
 
-var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 290, 304, 313, 326, 334, 342, 356, 375, 395, 410, 427, 443, 457, 469, 480, 490}
+var _apiMethod_index = [...]uint16{0, 17, 31, 45, 59, 82, 96, 109, 121, 141, 158, 173, 189, 210, 218, 227, 241, 249, 257, 277, 290, 304, 313, 326, 334, 342, 356, 375, 395, 410, 427, 443, 457, 469, 480, 490, 507, 526, 549, 562}
 
 func (i apiMethod) String() string {
 	if i < 0 || i >= apiMethod(len(_apiMethod_index)-1) {

boltdb/translate.go

@@ -12,7 +12,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/molecula/featurebase/v3"
+	pilosa "github.com/molecula/featurebase/v3"
+	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/pkg/errors"
 	bolt "go.etcd.io/bbolt"
 
@@ -32,6 +33,8 @@ var (
 
 	bucketKeys = []byte("keys")
 	bucketIDs  = []byte("ids")
+	bucketFree = []byte("free")
+	freeKey    = []byte("free")
 )
 
 const (
@@ -119,6 +122,8 @@ func (s *TranslateStore) Open() (err error) {
 			return err
 		} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
 			return err
+		} else if _, err := tx.CreateBucketIfNotExists(bucketFree); err != nil {
+			return err
 		}
 		return nil
 	}); err != nil {
@@ -230,14 +235,26 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) {
 			if idBucket == nil {
 				return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs)
 			}
+			freeBucket := tx.Bucket(bucketFree)
+			if freeBucket == nil {
+				return errors.Errorf(errFmtTranslateBucketNotFound, bucketFree)
+			}
 			puts := 0
+
+			// we create a freeIDGetter to reduce marshalling
+			getter := newFreeIDGetter(freeBucket)
+			defer getter.Close()
+
 			for idx, key := range keys {
 				id, boltKey := findIDByKey(keyBucket, key)
 				if id != 0 {
 					result[key] = id
 					continue
 				}
-				id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
+				// see if we can re-use any IDs first
+				if id = getter.GetFreeID(); id == 0 {
+					id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN)
+				}
 				idBytes := idScratch[puts*8 : puts*8+8]
 				binary.BigEndian.PutUint64(idBytes, id)
 				puts++
@@ -498,6 +515,88 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
 	}
 }
 
+type boltWrapper struct {
+	tx *bolt.Tx
+	db *bolt.DB
+}
+
+func (w *boltWrapper) Commit() error {
+	if w.tx != nil {
+		return w.tx.Commit()
+	}
+	return nil
+}
+
+func (w *boltWrapper) Rollback() {
+	if w.tx != nil {
+		w.tx.Rollback()
+	}
+}
+func (s *TranslateStore) FreeIDs() (*roaring.Bitmap, error) {
+	result := roaring.NewBitmap()
+	err := s.db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket(bucketFree)
+		if bkt == nil {
+			return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
+		}
+		b := bkt.Get(freeKey)
+		err := result.UnmarshalBinary(b)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	return result, err
+}
+func (s *TranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
+	bkt := tx.Bucket(bucketFree)
+	b := bkt.Get(freeKey)
+	buf := new(bytes.Buffer)
+	if b != nil { //if existing combine with newIDs
+		before := roaring.NewBitmap()
+		err := before.UnmarshalBinary(b)
+		if err != nil {
+			return err
+		}
+		final := newIDs.Union(before)
+		_, err = final.WriteTo(buf)
+		if err != nil {
+			return err
+		}
+	} else {
+		newIDs.WriteTo(buf)
+	}
+	return bkt.Put(freeKey, buf.Bytes())
+}
+
+// Delete removes the lookeup pairs in order to make avialble for reuse but doesn't commit the
+// transaction for that is tied to the associated rbf transaction being successful
+func (s *TranslateStore) Delete(records *roaring.Bitmap) (pilosa.Commitor, error) {
+	tx, err := s.db.Begin(true)
+	if err != nil {
+		return nil, err
+	}
+	keyBucket := tx.Bucket(bucketKeys)
+	idBucket := tx.Bucket(bucketIDs)
+	ids := records.Slice()
+	for i := range ids {
+		id := u64tob(ids[i])
+		boltKey := idBucket.Get(id)
+		err = keyBucket.Delete(boltKey)
+		if err != nil {
+			tx.Rollback()
+			return &boltWrapper{}, err
+		}
+		err = idBucket.Delete(id)
+		if err != nil {
+			tx.Rollback()
+			return &boltWrapper{}, err
+		}
+
+	}
+	return &boltWrapper{tx: tx}, s.MergeFree(tx, records)
+}
+
 // emptyKey is a sentinel byte slice which stands for "" as a key.
 var emptyKey = []byte{
 	0x00, 0x00, 0x00,
@@ -521,6 +620,84 @@ func findIDByKey(bkt *bolt.Bucket, key string) (uint64, []byte) {
 	return 0, boltKey
 }
 
+// freeIDGetter reduces the amount of marshaling required to get multiple ids
+type freeIDGetter struct {
+	freeBucket *bolt.Bucket
+	b          *roaring.Bitmap
+	changed    bool
+}
+
+// newFreeIDGetter initializes a new freeIDGetter. If at any point there is a
+// failure, it returns an error.
+//
+// NOTE: For changes to be persisted to the bucket, you must call
+// (*freeIDGetter).Close()
+func newFreeIDGetter(freeBucket *bolt.Bucket) *freeIDGetter {
+	g := &freeIDGetter{
+		freeBucket: freeBucket,
+	}
+	// we ignore this value because it's okay if we dont have a bitmap just yet
+	_ = g.getBitmap()
+	return g
+}
+
+func (g *freeIDGetter) getBitmap() bool {
+	if g.b == nil {
+		// get the bitmap from freeBucket
+		value := g.freeBucket.Get(freeKey)
+		if value == nil {
+			return false
+		}
+		// turn the value into a bitmap
+		b := roaring.NewBitmap()
+		if err := b.UnmarshalBinary(value); err != nil {
+			return false
+		}
+		g.b = b
+	}
+	return true
+}
+
+// GetFreeID tries to get a free ID from the free id bucket. If at any point it
+// fails to do so, it returns a 0. Otherwise, it returns the first free ID in the
+// bucket
+func (g *freeIDGetter) GetFreeID() (id uint64) {
+	if !g.getBitmap() {
+		return 0
+	}
+	// get the first free id
+	id, ok := g.b.Min()
+	if !ok {
+		return 0
+	}
+	// remove that id from the free id bitmap
+	if changed, err := g.b.RemoveN(id); changed == 0 || err != nil {
+		return 0
+	} else {
+		g.changed = true
+	}
+	return id
+}
+
+// Close persists any changes to the bitmap back to the bucket and then nils the
+// references for safety.
+func (g *freeIDGetter) Close() error {
+	if g.changed {
+		// convert bitmap to binary
+		buf, err := g.b.MarshalBinary()
+		if err != nil {
+			return errors.Wrap(err, "closing free ID Getter")
+		}
+		// put updated bitmap back into the freeBucket
+		if err := g.freeBucket.Put(freeKey, buf); err != nil {
+			return errors.Wrap(err, "closing free ID Getter")
+		}
+	}
+	g.b = nil
+	g.freeBucket = nil
+	return nil
+}
+
 func findKeyByID(bkt *bolt.Bucket, id uint64) string {
 	boltKey := bkt.Get(u64tob(id))
 	if bytes.Equal(boltKey, emptyKey) {

boltdb/translate_internal_test.go

@@ -0,0 +1,107 @@
+package boltdb
+
+import (
+	"path/filepath"
+	"testing"
+
+	"github.com/molecula/featurebase/v3/roaring"
+	bolt "go.etcd.io/bbolt"
+)
+
+func TestGetFreeID(t *testing.T) {
+	boltDir := t.TempDir()
+	db, err := bolt.Open(filepath.Join(boltDir, "testDB"), 0600, nil)
+	if err != nil {
+		t.Fatalf("unexpected error opening test boltdb: %v", err)
+	}
+	defer db.Close()
+
+	makeTestBucket := func(tx *bolt.Tx, b *roaring.Bitmap) *bolt.Bucket {
+		if b == nil {
+			t.Fatalf("unexpected nil bitmap")
+		}
+		free, err := tx.CreateBucketIfNotExists(bucketFree)
+		if err != nil {
+			t.Fatalf("unexpected error making freeBucket: %v", err)
+		}
+		buf, err := b.MarshalBinary()
+		if err != nil {
+			t.Fatalf("unexpected error marshaling bitmap (%v) to binary: %v", b, err)
+		}
+		if err := free.Put(freeKey, buf); err != nil {
+			t.Fatalf("unexpected error adding data (%v) to freeBucket: %v", b, err)
+		}
+		return free
+	}
+
+	for name, test := range map[string]struct {
+		bits *roaring.Bitmap
+		want uint64
+	}{
+		"bucket is there, but nobody's home": {
+			bits: roaring.NewBitmap(),
+			want: 0,
+		},
+		"good bucket": {
+			bits: roaring.NewBitmap(1, 2, 34, 55, 9000),
+			want: 1,
+		},
+	} {
+		t.Run(name, func(t *testing.T) {
+			tx, err := db.Begin(true)
+			if err != nil {
+				t.Fatalf("unexpected error starting bolt transaction: %v", err)
+			}
+			defer tx.Rollback()
+			freeBucket := makeTestBucket(tx, test.bits)
+
+			getter := newFreeIDGetter(freeBucket)
+			defer getter.Close()
+			if got := getter.GetFreeID(); got != test.want {
+				t.Fatalf("expected %v got %v", test.want, got)
+			}
+		})
+	}
+
+	t.Run("CorrectOrdering", func(t *testing.T) {
+		tx, err := db.Begin(true)
+		if err != nil {
+			t.Fatalf("unexpected error starting bolt transaction: %v", err)
+		}
+		defer tx.Rollback()
+
+		bucket := makeTestBucket(tx, roaring.NewBitmap(1, 34, 2, 55, 9000))
+
+		getter := newFreeIDGetter(bucket)
+		defer getter.Close()
+		for _, want := range []uint64{1, 2, 34, 55, 9000} {
+			if got := getter.GetFreeID(); got != want {
+				t.Fatalf("expected %v got %v", want, got)
+			}
+		}
+		if got := getter.GetFreeID(); got != 0 {
+			t.Fatalf("expected 0 got %v", got)
+		}
+	})
+
+	t.Run("NotABitmap", func(t *testing.T) {
+		tx, err := db.Begin(true)
+		if err != nil {
+			t.Fatalf("unexpected error starting bolt transaction: %v", err)
+		}
+		defer tx.Rollback()
+
+		free, err := tx.CreateBucketIfNotExists(bucketFree)
+		if err != nil {
+			t.Fatalf("unexpected error making freeBucket: %v", err)
+		}
+		if err := free.Put(freeKey, []byte("this isn't right!")); err != nil {
+			t.Fatalf("unexpected error adding data to freeBucket: %v", err)
+		}
+		getter := newFreeIDGetter(free)
+		defer getter.Close()
+		if got := getter.GetFreeID(); got != 0 {
+			t.Fatalf("expected 0 got %v", got)
+		}
+	})
+}

boltdb/translate_test.go

@@ -10,8 +10,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v3"
+	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/boltdb"
+	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/topology"
 )
@@ -385,7 +386,53 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
 	s.Path = f.Name()
 	return s
 }
+func TestTranslateStore_Delete(t *testing.T) {
+	s := MustOpenNewTranslateStore(t)
+	defer MustCloseTranslateStore(s)
 
+	// Setup initial keys.
+	ids, err := s.CreateKeys("foo", "bar", "deleteme")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	records := roaring.NewBitmap(ids["deleteme"])
+	c, err := s.Delete(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = c.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	r, e := s.FreeIDs()
+	if e != nil {
+		t.Fatal(err)
+	}
+	freeids := r.Slice()
+	if len(freeids) == 0 {
+		t.Fatalf("expected to have free id")
+	}
+	if freeids[0] != ids["deleteme"] {
+		t.Fatalf("expected [%v] and got %v", ids["deleteme"], freeids[0])
+	}
+
+	records2 := roaring.NewBitmap(ids["foo"])
+	c, err = s.Delete(records2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = c.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	r, e = s.FreeIDs()
+	if e != nil {
+		t.Fatal(err)
+	}
+	freeids = r.Slice()
+	if len(freeids) != 2 {
+		t.Fatalf("expected to have 2 free ids")
+	}
+}
 func TestTranslateStore_ReadWrite(t *testing.T) {
 	t.Run("WriteTo_ReadFrom", func(t *testing.T) {
 		s := MustOpenNewTranslateStore(t)
@@ -408,7 +455,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
 
 		// Put the contents of the store into a buffer.
 		buf := bytes.NewBuffer(nil)
-		expN := int64(32768)
+		expN := s.Size()
 
 		// After this, the buffer should contain batch0.
 		if n, err := s.WriteTo(buf); err != nil {
@@ -458,7 +505,7 @@ func TestTranslateStore_ReadWrite(t *testing.T) {
 func MustOpenNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
 	s := MustNewTranslateStore(tb)
 	if err := s.Open(); err != nil {
-		panic(err)
+		tb.Fatalf("opening s: %v", err)
 	}
 	return s
 }

cache.go

@@ -194,6 +194,14 @@ func (c *rankCache) BulkAdd(id uint64, n uint64) {
 	}
 
 	c.entries[id] = n
+
+	// FB-1206: Periodically invalidate the cache when we are bulk loading
+	// as this can take up an upbounded amount of memory. This is especially
+	// true when restoring shards as all rows will be added.
+	if len(c.entries) > int(2*c.maxEntries) {
+		c.stats.Count(MetricRecalculateCache, 1, 1.0)
+		c.recalculate()
+	}
 }
 
 // Get returns a count for a given id.

cache_test.go

@@ -70,3 +70,15 @@ func TestCache_Rank_Dirty(t *testing.T) {
 		t.Fatalf("wrote %v but got %v", expect, got)
 	}
 }
+
+func TestCache_Rank_BulkAdd(t *testing.T) {
+	const cacheSize = 10
+	cache := pilosa.NewRankCache(uint32(cacheSize))
+
+	for i := uint64(0); i < 1000; i++ {
+		cache.BulkAdd(i, i)
+		if n := cache.Len(); n > cacheSize*2 {
+			t.Fatalf("entry count exceed 2x cache size: %d", n)
+		}
+	}
+}

catcher.go

@@ -26,6 +26,11 @@ func init() {
 
 var _ Tx = (*catcherTx)(nil)
 
+func (c *catcherTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
+	c.b.RemoveChannel(index, field, view, shard, a, resChan)
+	return
+}
+
 func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
 	return c.b.NewTxIterator(index, field, view, shard)
 }

client/client.go

@@ -832,31 +832,24 @@ func (c *Client) httpRequest(method string, path string, data []byte, headers ma
 		body   []byte
 		err    error
 	)
-	// try at most maxHosts non-failed hosts; protect against broken cluster.removeHost
-	for i := 0; i < maxHosts; i++ {
+	// try request on host, if it fails, try again on primary
+	for i := 0; i <= 1; i++ {
 		host, herr := c.host(usePrimary)
 		if herr != nil {
 			return status, nil, errors.Wrapf(herr, "getting host, previous err: %v", err)
 		}
 		// doRequest implements expotential backoff
 		status, body, err = c.doRequest(host, method, path, c.augmentHeaders(headers), data)
-		if err == nil {
+		// conditions when primary should not be tried
+		if err == nil || usePrimary || path == "/status" {
 			break
 		}
-		if c.manualServerURI == nil {
-			if usePrimary {
-				c.primaryLock.Lock()
-				c.primaryURI = nil
-				c.primaryLock.Unlock()
-			} else {
-				c.logger.Printf("removing host (%s) due to '%v'\n", host.Normalize(), err)
-				c.cluster.RemoveHost(host)
-			}
-		}
+
+		usePrimary = true
 	}
 
 	if err != nil {
-		err = errors.Wrap(err, ErrTriedMaxHosts.Error())
+		err = errors.Wrap(err, ErrHTTPRequest.Error())
 	}
 
 	return status, body, err
@@ -1674,17 +1667,18 @@ type SchemaField struct {
 
 // SchemaOptions contains options for a field or an index.
 type SchemaOptions struct {
-	FieldType      FieldType   `json:"type"`
-	CacheType      string      `json:"cacheType"`
-	CacheSize      uint        `json:"cacheSize"`
-	TimeQuantum    string      `json:"timeQuantum"`
-	Min            pql.Decimal `json:"min"`
-	Max            pql.Decimal `json:"max"`
-	Scale          int64       `json:"scale"`
-	Keys           bool        `json:"keys"`
-	NoStandardView bool        `json:"noStandardView"`
-	TrackExistence bool        `json:"trackExistence"`
-	TimeUnit       string      `json:"timeUnit"`
+	FieldType      FieldType     `json:"type"`
+	CacheType      string        `json:"cacheType"`
+	CacheSize      uint          `json:"cacheSize"`
+	TimeQuantum    string        `json:"timeQuantum"`
+	Ttl            time.Duration `json:"ttl"`
+	Min            pql.Decimal   `json:"min"`
+	Max            pql.Decimal   `json:"max"`
+	Scale          int64         `json:"scale"`
+	Keys           bool          `json:"keys"`
+	NoStandardView bool          `json:"noStandardView"`
+	TrackExistence bool          `json:"trackExistence"`
+	TimeUnit       string        `json:"timeUnit"`
 }
 
 func (so SchemaOptions) asIndexOptions() *IndexOptions {
@@ -1702,6 +1696,7 @@ func (so SchemaOptions) asFieldOptions() *FieldOptions {
 		cacheSize:      int(so.CacheSize),
 		cacheType:      CacheType(so.CacheType),
 		timeQuantum:    TimeQuantum(so.TimeQuantum),
+		ttl:            so.Ttl,
 		min:            so.Min,
 		max:            so.Max,
 		scale:          so.Scale,

client/client_it_test.go

@@ -497,7 +497,7 @@ func TestClientAgainstCluster(t *testing.T) {
 				tmpcli, _ := NewClient(NewClusterWithHost(uri, uri, uri, uri), OptClientRetries(0))
 
 				_, err := tmpcli.Query(testIndex.All())
-				require.Error(t, err, ErrTriedMaxHosts)
+				require.Error(t, err, ErrHTTPRequest)
 			})
 
 			t.Run("InvalidQuery", func(t *testing.T) {

client/cluster.go

@@ -65,18 +65,6 @@ func (c *Cluster) Host() *pnet.URI {
 	return host
 }
 
-// RemoveHost black lists the host with the given pnet.URI from the cluster.
-func (c *Cluster) RemoveHost(address *pnet.URI) {
-	c.mutex.Lock()
-	defer c.mutex.Unlock()
-	for i, uri := range c.hosts {
-		if uri.Equals(address) {
-			c.okList[i] = false
-			break
-		}
-	}
-}
-
 // Hosts returns all available hosts in the cluster.
 func (c *Cluster) Hosts() []pnet.URI {
 	c.mutex.RLock()

client/cluster_test.go

@@ -49,22 +49,3 @@ func TestHosts(t *testing.T) {
 		t.Fatalf("Host should return a value if there are hosts in the cluster")
 	}
 }
-
-func TestRemoveHost(t *testing.T) {
-	uri, err := pnet.NewURIFromAddress("index1.pilosa.com:9999")
-	if err != nil {
-		t.Fatal(err)
-	}
-	c := NewClusterWithHost(uri)
-	if len(c.hosts) != 1 {
-		t.Fatalf("The cluster should contain the host")
-	}
-	uri, err = pnet.NewURIFromAddress("index1.pilosa.com:9999")
-	if err != nil {
-		t.Fatal(err)
-	}
-	c.RemoveHost(uri)
-	if len(c.Hosts()) != 0 {
-		t.Fatalf("The cluster should not contain the host")
-	}
-}

client/error.go

@@ -12,7 +12,7 @@ var (
 	ErrInvalidFieldName            = errors.New("Invalid field name")
 	ErrInvalidLabel                = errors.New("Invalid label")
 	ErrInvalidKey                  = errors.New("Invalid key")
-	ErrTriedMaxHosts               = errors.New("Tried max hosts, still failing")
+	ErrHTTPRequest                 = errors.New("Failed all HTTP retries")
 	ErrAddrURIClusterExpected      = errors.New("Addresses, URIs or a cluster is expected")
 	ErrInvalidQueryOption          = errors.New("Invalid query option")
 	ErrInvalidIndexOption          = errors.New("Invalid index option")

client/ingest_api_batch_test.go

@@ -133,7 +133,6 @@ func TestIngestAPIBatchAdd(t *testing.T) {
 }
 
 func TestIngestAPIBatch(t *testing.T) {
-	t.Skip("causing sporadic CI failures... on my list to debug, but this code doesn't affect anyone's production anyhow (jaffee)")
 	c := test.MustRunCluster(t, 3)
 	defer c.Close()
 

client/orm.go

@@ -728,6 +728,7 @@ type FieldInfo struct {
 type FieldOptions struct {
 	fieldType      FieldType
 	timeQuantum    TimeQuantum
+	ttl            time.Duration
 	cacheType      CacheType
 	cacheSize      int
 	min            pql.Decimal
@@ -751,6 +752,11 @@ func (fo FieldOptions) TimeQuantum() TimeQuantum {
 	return fo.timeQuantum
 }
 
+// Ttl returns the configured ttl for a time field.
+func (fo FieldOptions) Ttl() time.Duration {
+	return fo.ttl
+}
+
 // CacheType returns the configured cache type for a "set" field. Empty string
 // otherwise.
 func (fo FieldOptions) CacheType() CacheType {
@@ -826,6 +832,7 @@ func (fo FieldOptions) String() string {
 	case FieldTypeTime:
 		mopt["timeQuantum"] = string(fo.timeQuantum)
 		mopt["noStandardView"] = fo.noStandardView
+		mopt["ttl"] = fo.ttl.String()
 	case FieldTypeTimestamp:
 		mopt["min"] = fo.min
 		mopt["max"] = fo.max
@@ -913,6 +920,12 @@ func OptFieldTypeTime(quantum TimeQuantum, opts ...bool) FieldOption {
 	}
 }
 
+func OptFieldTtl(dur time.Duration) FieldOption {
+	return func(options *FieldOptions) {
+		options.ttl = dur
+	}
+}
+
 // Timestamp field range.
 var (
 	DefaultEpoch = time.Unix(0, 0).UTC() // 1970-01-01T00:00:00Z

client/orm_test.go

@@ -925,7 +925,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(0, 0),
 			pql.NewDecimal(0, 0),
 			"",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("IntFieldOptions", func(t *testing.T) {
@@ -944,7 +945,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(-10, 0),
 			pql.NewDecimal(100, 0),
 			"",
-			"")
+			"",
+			0)
 
 		field = sampleIndex.Field("int-field2", OptFieldTypeInt(-10))
 		jsonString = field.options.String()
@@ -962,7 +964,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(-10, 0),
 			pql.NewDecimal(math.MaxInt64, 0),
 			"",
-			"")
+			"",
+			0)
 		field = sampleIndex.Field("int-field3", OptFieldTypeInt())
 		jsonString = field.options.String()
 		targetString = fmt.Sprintf(`{"options":{"type":"int","min":%d,"max":%d}}`, math.MinInt64, math.MaxInt64)
@@ -978,7 +981,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(math.MinInt64, 0),
 			pql.NewDecimal(math.MaxInt64, 0),
 			"",
-			"")
+			"",
+			0)
 
 		field = sampleIndex.Field("int-field4", OptFieldTypeInt(), OptFieldForeignIndex("blerg"))
 		jsonString = field.options.String()
@@ -995,7 +999,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(math.MinInt64, 0),
 			pql.NewDecimal(math.MaxInt64, 0),
 			"blerg",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("TimeFieldOptions", func(t *testing.T) {
@@ -1004,7 +1009,7 @@ func TestORM(t *testing.T) {
 			t.Fatalf("field noStandardView %v != %v", true, field.Opts().NoStandardView())
 		}
 		jsonString := field.options.String()
-		targetString := `{"options":{"noStandardView":true,"type":"time","timeQuantum":"DH"}}`
+		targetString := `{"options":{"noStandardView":true,"type":"time","timeQuantum":"DH","ttl":"0s"}}`
 		if sortedString(targetString) != sortedString(jsonString) {
 			t.Fatalf("`%s` != `%s`", targetString, jsonString)
 		}
@@ -1017,7 +1022,31 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(0, 0),
 			pql.NewDecimal(0, 0),
 			"",
-			"")
+			"",
+			0)
+	})
+
+	t.Run("TtlOptions", func(t *testing.T) {
+		field := sampleIndex.Field("ttl-field", OptFieldTypeTime(TimeQuantumDayHour, true), OptFieldTtl(0))
+		if true != field.Opts().NoStandardView() {
+			t.Fatalf("field noStandardView %v != %v", true, field.Opts().NoStandardView())
+		}
+		jsonString := field.options.String()
+		targetString := `{"options":{"noStandardView":true,"type":"time","timeQuantum":"DH","ttl":"0s"}}`
+		if sortedString(targetString) != sortedString(jsonString) {
+			t.Fatalf("`%s` != `%s`", targetString, jsonString)
+		}
+		compareFieldOptions(t,
+			field.Options(),
+			FieldTypeTime,
+			TimeQuantumDayHour,
+			CacheTypeDefault,
+			0,
+			pql.NewDecimal(0, 0),
+			pql.NewDecimal(0, 0),
+			"",
+			"",
+			0)
 	})
 
 	t.Run("MutexFieldOptions", func(t *testing.T) {
@@ -1036,7 +1065,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(0, 0),
 			pql.NewDecimal(0, 0),
 			"",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("BoolFieldOptions", func(t *testing.T) {
@@ -1055,7 +1085,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(0, 0),
 			pql.NewDecimal(0, 0),
 			"",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("DecimalFieldOptions", func(t *testing.T) {
@@ -1074,7 +1105,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(7, 3),
 			pql.NewDecimal(999, 3),
 			"",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("DecimalFieldOptions", func(t *testing.T) {
@@ -1093,7 +1125,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(7, 3),
 			pql.NewDecimal(999, 3),
 			"",
-			"")
+			"",
+			0)
 	})
 
 	t.Run("TimestampFieldOptions", func(t *testing.T) {
@@ -1114,7 +1147,8 @@ func TestORM(t *testing.T) {
 			pql.NewDecimal(MinTimestamp.UnixNano()/TimeUnitNanos(pilosa.TimeUnitSeconds), 0),
 			pql.NewDecimal(MaxTimestamp.UnixNano()/TimeUnitNanos(pilosa.TimeUnitSeconds), 0),
 			"",
-			pilosa.TimeUnitSeconds)
+			pilosa.TimeUnitSeconds,
+			0)
 
 	})
 
@@ -1163,7 +1197,7 @@ func comparePQL(t *testing.T, target string, q PQLQuery) {
 	}
 }
 
-func compareFieldOptions(t *testing.T, opts *FieldOptions, fieldType FieldType, timeQuantum TimeQuantum, cacheType CacheType, cacheSize int, min pql.Decimal, max pql.Decimal, foreignIndex string, timeUnit string) {
+func compareFieldOptions(t *testing.T, opts *FieldOptions, fieldType FieldType, timeQuantum TimeQuantum, cacheType CacheType, cacheSize int, min pql.Decimal, max pql.Decimal, foreignIndex string, timeUnit string, ttl time.Duration) {
 	if fieldType != opts.Type() {
 		t.Fatalf("%s != %s", fieldType, opts.Type())
 	}
@@ -1188,6 +1222,9 @@ func compareFieldOptions(t *testing.T, opts *FieldOptions, fieldType FieldType,
 	if timeUnit != opts.TimeUnit() {
 		t.Fatalf("%s != %s", timeUnit, opts.TimeUnit())
 	}
+	if ttl != opts.Ttl() {
+		t.Fatalf("%s != %s", ttl, opts.Ttl())
+	}
 }
 
 func sortedString(s string) string {

cmd/import.go

@@ -46,6 +46,7 @@ omitted. If it is present then its format should be YYYY-MM-DDTHH:MM.
 	flags.StringVar(&Importer.FieldOptions.CacheType, "field-cache-type", pilosa.CacheTypeRanked, "Specify the cache type for a set field on creation. One of: none, lru, ranked")
 	flags.Uint32Var(&Importer.FieldOptions.CacheSize, "field-cache-size", 50000, "Specify the cache size for a set field on creation")
 	flags.Var(&Importer.FieldOptions.TimeQuantum, "field-time-quantum", "Specify the time quantum for a time field on creation. One of: D, DH, H, M, MD, MDH, Y, YM, YMD, YMDH")
+	flags.DurationVarP(&Importer.FieldOptions.Ttl, "time-to-live", "t", 0, "Specify the time to live for views created by time quantum. Supported time unit: \"s\", \"m\", \"h\"") // \"ns\", \"us\" (or \"µs\"), \"ms\" also supported but ommitted for simplicity
 	flags.IntVarP(&Importer.BufferSize, "buffer-size", "s", 10000000, "Number of bits to buffer/sort before importing.")
 	flags.BoolVarP(&Importer.Sort, "sort", "", false, "Enables sorting before import.")
 	flags.BoolVarP(&Importer.CreateSchema, "create", "e", false, "Create the schema if it does not exist before import.")

cmd/roaring-migrate/main.go

@@ -4,6 +4,7 @@ package main
 import (
 	"encoding/json"
 	"fmt"
+	"io"
 	"io/ioutil"
 	"os"
 	"path/filepath"
@@ -13,25 +14,42 @@ import (
 	"syscall"
 
 	pilosa "github.com/molecula/featurebase/v3"
+	"github.com/molecula/featurebase/v3/logger"
 	"github.com/molecula/featurebase/v3/rbf"
 	"github.com/molecula/featurebase/v3/rbf/cfg"
 	"github.com/molecula/featurebase/v3/roaring"
 	txkey "github.com/molecula/featurebase/v3/short_txkey"
+	"github.com/molecula/featurebase/v3/vprint"
 	"github.com/spf13/cobra"
 )
 
+var visited map[string]int64
+var glogger = logger.NewStandardLogger(os.Stdout)
+
+const (
+	Version = "1.0"
+)
+
 func main() {
+	os.Exit(realMain())
+}
+func realMain() int {
+	visited = make(map[string]int64)
 	var dataDir, backupPath string
+	var verbose bool
 	cmdMigrate := &cobra.Command{
 		Use:   "roaring-migrate",
 		Short: "convert roaring pilosa backup to rbf",
 		Long:  `roaring-migrate uses the pilosa data-dir for each node, and produces a new backup that is able to be restored from utilizing the new pilosa restore tool.`,
 		Run: func(cmd *cobra.Command, args []string) {
+			if verbose {
+				glogger.Infof("Version: %v", Version)
+			}
 			nodes := strings.Split(dataDir, ",")
 			for _, nodePath := range nodes {
-				err := Migrate(nodePath, backupPath)
+				err := Migrate(nodePath, backupPath, verbose)
 				if err != nil {
-					fmt.Println("Error", err)
+					glogger.Errorf("%v", Version)
 					return
 				}
 
@@ -40,24 +58,24 @@ func main() {
 	}
 	cmdMigrate.Flags().StringVarP(&dataDir, "data-dir", "d", "", "source directories for each node seperated by commas")
 	cmdMigrate.Flags().StringVarP(&backupPath, "backup-dir", "b", "", "location of backup directory")
+	cmdMigrate.Flags().BoolVar(&verbose, "verbose", false, "additional progress information")
 	err := cmdMigrate.MarkFlagRequired("data-dir")
 	if err != nil {
-		fmt.Println("Error setting flag data-dir")
-		os.Exit(1)
-		return
+		glogger.Errorf("Error setting flag data-dir")
+		return 1
 	}
 	err = cmdMigrate.MarkFlagRequired("backup-dir")
 	if err != nil {
-		fmt.Println("Error setting flag backup-dir")
-		os.Exit(1)
-		return
+		glogger.Errorf("Error setting flag backup-dir")
+		return 1
 	}
 
 	err = cmdMigrate.Execute()
 	if err != nil {
-		fmt.Println("exec error", err)
-		os.Exit(1)
+		glogger.Errorf("exec error %v", err)
+		return 1
 	}
+	return 0
 }
 
 func FetchFragments(base string) []string {
@@ -67,7 +85,7 @@ func FetchFragments(base string) []string {
 
 		// first thing to do, check error. and decide what to do about it
 		if errX != nil {
-			fmt.Printf("error 「%v」 at a path 「%q」\n", errX, pathX)
+			glogger.Errorf("error 「%v」 at a path 「%q」\n", errX, pathX)
 			return errX
 		}
 		pathX = pathX[len(base):]
@@ -82,7 +100,7 @@ func FetchFragments(base string) []string {
 	err := filepath.Walk(base, ff)
 
 	if err != nil {
-		fmt.Printf("error walking the path %q: %v\n", base, err)
+		glogger.Errorf("error walking the path %q: %v\n", base, err)
 	}
 	return fragments
 }
@@ -94,6 +112,14 @@ type local struct {
 	Fields    []*pilosa.FieldInfo `json:"fields,omitempty"`
 }
 
+func fileExists(filename string) (bool, int64) {
+	info, err := os.Stat(filename)
+	if os.IsNotExist(err) {
+		return false, 0
+	}
+	return !info.IsDir(), info.Size()
+}
+
 func BuildSchema(dataDir string) ([]byte, error) {
 	//need to find all the ".meta" files and load as field options
 
@@ -105,7 +131,7 @@ func BuildSchema(dataDir string) ([]byte, error) {
 
 		// first thing to do, check error. and decide what to do about it
 		if errX != nil {
-			fmt.Printf("error 「%v」 at a path 「%q」\n", errX, pathX)
+			glogger.Infof("error 「%v」 at a path 「%q」\n", errX, pathX)
 			return errX
 		}
 		pathX = pathX[len(dataDir):]
@@ -115,7 +141,7 @@ func BuildSchema(dataDir string) ([]byte, error) {
 			if strings.Contains(pathX, ".meta") {
 				//convert the file to a fieldOptions
 				// ex: metaPath /trait_store/aba/.meta
-				fmt.Println("PATHX", pathX)
+				glogger.Infof("PATHX %v", pathX)
 				t := strings.Split(pathX, "/")
 				index := t[1]
 				src := dataDir + pathX
@@ -160,7 +186,7 @@ func BuildSchema(dataDir string) ([]byte, error) {
 	err := filepath.Walk(dataDir, ff)
 
 	if err != nil {
-		fmt.Printf("error walking the path %q: %v\n", dataDir, err)
+		glogger.Errorf("error walking the path %q: %v\n", dataDir, err)
 	}
 	return json.MarshalIndent(schemaSerializer, "", "    ")
 }
@@ -183,7 +209,7 @@ func (d *rbfFile) getDB(path, index string, shard uint64) (*rbf.DB, error) {
 	if d.last != src {
 		d.Close()
 		d.last = src
-		fmt.Println("RBF:", src)
+		glogger.Infof("RBF: %v", src)
 		c := cfg.NewDefaultConfig()
 		c.FsyncEnabled = false
 		c.MinWALCheckpointSize = 0
@@ -197,20 +223,34 @@ func (d *rbfFile) getDB(path, index string, shard uint64) (*rbf.DB, error) {
 	return d.working, nil
 }
 func (d *rbfFile) Close() error {
+	defer func() error {
+		// clean up the temp directory
+		err := os.RemoveAll(d.temp)
+		if err != nil {
+			return err
+		}
+		return nil
+	}()
+
 	if d.last != "" {
 		d.working.Close()
+
 		//if d.last exists only keep the biggest
-		err := os.MkdirAll(filepath.Dir(d.last), 0777)
-		if err != nil {
-			return err
+		exists, sz := fileExists(d.last)
+		src := filepath.Join(d.temp, "data")
+		if !exists {
+			err := os.MkdirAll(filepath.Dir(d.last), 0777)
+			if err != nil {
+				return err
+			}
+		} else {
+			_, sz2 := fileExists(src)
+			if sz > sz2 {
+				return nil
+			}
 		}
 		// move the datafile backup shard
-		err = os.Rename(filepath.Join(d.temp, "data"), d.last)
-		if err != nil {
-			return err
-		}
-		//cleanup the tempdirectory
-		err = os.RemoveAll(d.temp)
+		err := os.Rename(src, d.last)
 		if err != nil {
 			return err
 		}
@@ -218,19 +258,26 @@ func (d *rbfFile) Close() error {
 	return nil
 }
 func copyFile(src, dest string) error {
-	input, err := ioutil.ReadFile(src)
+	from, err := os.Open(src)
 	if err != nil {
 		return err
 	}
+	defer from.Close()
 
-	err = ioutil.WriteFile(dest, input, 0644)
+	to, err := os.OpenFile(dest, os.O_RDWR|os.O_CREATE, 0644)
+	if err != nil {
+		return err
+	}
+	defer to.Close()
+
+	_, err = io.Copy(to, from)
 	if err != nil {
 		return err
 	}
 	return nil
 }
 
-func Migrate(dataDir, backupPath string) error {
+func Migrate(dataDir, backupPath string, verbose bool) error {
 	dataDir = strings.TrimSuffix(dataDir, "/")
 
 	err := os.MkdirAll(backupPath, 0777)
@@ -279,7 +326,21 @@ func Migrate(dataDir, backupPath string) error {
 	bm := roaring.NewSliceBitmap()
 	for _, filename := range raw {
 		index, field, view, shard := Extract(filename)
-
+		sz, before := visited[filename]
+		fi, _ := os.Stat(dataDir + filename)
+		if field != "_exists" {
+			if !before {
+				visited[filename] = fi.Size()
+			} else {
+				if fi.Size() <= sz {
+					continue // skip it
+				}
+				visited[filename] = fi.Size()
+			}
+		}
+		if verbose {
+			glogger.Infof("processing: %v", dataDir+filename)
+		}
 		content, err := ioutil.ReadFile(dataDir + filename)
 		if err != nil {
 			return err
@@ -293,35 +354,19 @@ func Migrate(dataDir, backupPath string) error {
 		if err != nil {
 			return err
 		}
-		tx, err := db.Begin(true)
-		if err != nil {
-			return err
-		}
 		key := string(txkey.Prefix(index, field, view, shard))
-		itr, ok := bm.Containers.Iterator(0)
-		if ok {
-			for itr.Next() {
-				k, v := itr.Value()
-				tx.PutContainer(key, k, v)
-
-			}
-		}
+		tx, err := db.Begin(true)
+		tx.AddRoaring(key, bm)
 		err = tx.Commit()
-		if err != nil {
-			return err
-		}
 	}
 	cache.Close()
 	keys := FetchIndexKeys(dataDir)
 	for _, filename := range keys {
-		fmt.Println("index keys", filename)
-		content, err := ioutil.ReadFile(filepath.Join(dataDir, filename))
-		if err != nil {
-			return err
-		}
+		glogger.Infof("index keys %v", filename)
+		srcFile := filepath.Join(dataDir, filename)
 		parts := strings.Split(filename, "/")
 		destFile := filepath.Join(backupPath, "indexes", parts[1], "translate", parts[3])
-		err = writeIfBigger(destFile, content)
+		err = writeIfBigger(destFile, srcFile)
 		if err != nil {
 			return err
 		}
@@ -330,14 +375,11 @@ func Migrate(dataDir, backupPath string) error {
 	//deal with index field(row)keys
 	keys = FetchRowkeys(dataDir)
 	for _, filename := range keys {
-		fmt.Println("field", filename)
-		content, err := ioutil.ReadFile(dataDir + filename)
-		if err != nil {
-			return err
-		}
+		glogger.Infof("field %v", filename)
+		srcFile := dataDir + filename
 		parts := strings.Split(filename, "/")
 		destFile := filepath.Join(backupPath, "indexes", parts[1], "fields", parts[2], "translate")
-		err = writeIfBigger(destFile, content)
+		err = writeIfBigger(destFile, srcFile)
 		if err != nil {
 			return err
 		}
@@ -345,16 +387,21 @@ func Migrate(dataDir, backupPath string) error {
 	return nil
 }
 
-func writeIfBigger(dst string, content []byte) error {
+func writeIfBigger(dst string, srcFile string) error {
 	if stats, err := os.Stat(dst); os.IsNotExist(err) {
 		err = os.MkdirAll(filepath.Dir(dst), 0777)
 		if err != nil {
 			return err
 		}
-		return ioutil.WriteFile(dst, content, 0644)
+		return copyFile(srcFile, dst)
 	} else {
-		if stats.Size() < int64(len(content)) {
-			return ioutil.WriteFile(dst, content, 0644)
+		stats2, err := os.Stat(srcFile)
+		if err != nil {
+			return err
+		}
+		if stats.Size() < stats2.Size() {
+			vprint.VV("Bigger %v %v", stats.Size(), stats2.Size())
+			return copyFile(srcFile, dst)
 		}
 	}
 	return nil //simply skip it
@@ -376,7 +423,7 @@ func FetchIndexKeys(base string) []string {
 
 		// first thing to do, check error. and decide what to do about it
 		if errX != nil {
-			fmt.Printf("error 「%v」 at a path 「%q」\n", errX, pathX)
+			glogger.Errorf("error 「%v」 at a path 「%q」\n", errX, pathX)
 			return errX
 		}
 		pathX = pathX[len(base):]
@@ -393,7 +440,7 @@ func FetchIndexKeys(base string) []string {
 	err := filepath.Walk(base, ff)
 
 	if err != nil {
-		fmt.Printf("error walking the path %q: %v\n", base, err)
+		glogger.Errorf("error walking the path %q: %v\n", base, err)
 	}
 	return directory
 }
@@ -405,7 +452,7 @@ func FetchRowkeys(base string) []string {
 
 		// first thing to do, check error. and decide what to do about it
 		if errX != nil {
-			fmt.Printf("error 「%v」 at a path 「%q」\n", errX, pathX)
+			glogger.Errorf("error 「%v」 at a path 「%q」\n", errX, pathX)
 			return errX
 		}
 		pathX = pathX[len(base):]
@@ -427,7 +474,7 @@ func FetchRowkeys(base string) []string {
 	err := filepath.Walk(base, ff)
 
 	if err != nil {
-		fmt.Printf("error walking the path %q: %v\n", base, err)
+		glogger.Errorf("error walking the path %q: %v\n", base, err)
 	}
 	return directory
 }

cmd/roaring-migrate/main_test.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"io/ioutil"
+	"os"
+	"testing"
+)
+
+func TestFileExists(t *testing.T) {
+	fileName := "missing"
+	if x, _ := fileExists(fileName); x {
+		t.Fatalf("file %v doesn't exist", fileName)
+	}
+	file, err := os.Create(fileName)
+	if err != nil {
+		t.Fatal(err)
+	}
+	file.Close()
+
+	if x, _ := fileExists(fileName); !x {
+		t.Fatalf("file %v doesn't exist", fileName)
+	}
+
+	t.Cleanup(func() {
+		os.Remove(fileName)
+	})
+}
+
+func TestMainProgram(t *testing.T) {
+	os.Args = []string{"roaring-migrate",
+		"--verbose",
+	}
+	if realMain() == 0 {
+		t.Fatal("should fail and it succeeded")
+	}
+	os.Args = []string{"roaring-migrate",
+		"--verbose",
+	}
+	if realMain() == 0 {
+		t.Fatal("should fail and it succeeded")
+	}
+	dir, err := ioutil.TempDir("", "backup")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer os.RemoveAll(dir) // clean up
+	os.Args = []string{"roaring-migrate",
+		"--verbose=true",
+		"--data-dir=testdata/data-dir/",
+		"--backup-dir=" + dir,
+	}
+	if realMain() == 1 {
+		t.Fatal("shouldn't fail")
+	}
+
+}

cmd/roaring-migrate/testdata/data-dir/.id

@@ -0,0 +1 @@
+6fc20f49-edf3-4211-8f6d-c670258ee6ea

cmd/roaring-migrate/testdata/data-dir/.startup.log

@@ -0,0 +1 @@
+2022-02-14T11:49:34.20065623-06:00	v2.7.0

cmd/roaring-migrate/testdata/data-dir/.topology

@@ -0,0 +1,2 @@
+
+$a317bd70-60ed-4723-99fa-3067563a708e$6fc20f49-edf3-4211-8f6d-c670258ee6ea

cmd/roaring-migrate/testdata/data-dir/repository/.meta

@@ -0,0 +1 @@
+