From cb21c105a8fdff906d14367dcb01d946392d8c15 Mon Sep 17 00:00:00 2001 From: bednar Date: Wed, 10 Oct 2018 09:54:05 +0200 Subject: [PATCH] Added to functions --- flux-dsl/README.md | 14 ++ .../main/java/org/influxdata/flux/Flux.java | 171 +++++++++++++- .../influxdata/flux/functions/CountFlux.java | 2 +- .../flux/functions/CovarianceFlux.java | 2 +- .../flux/functions/CumulativeSumFlux.java | 5 +- .../flux/functions/DerivativeFlux.java | 2 +- .../flux/functions/DifferenceFlux.java | 2 +- .../flux/functions/DistinctFlux.java | 4 +- .../influxdata/flux/functions/DropFlux.java | 5 +- .../flux/functions/DuplicateFlux.java | 4 +- .../influxdata/flux/functions/FilterFlux.java | 2 +- .../influxdata/flux/functions/FirstFlux.java | 4 +- .../influxdata/flux/functions/FromFlux.java | 4 +- .../influxdata/flux/functions/GroupFlux.java | 4 +- .../flux/functions/IntegralFlux.java | 2 +- .../influxdata/flux/functions/JoinFlux.java | 2 +- .../influxdata/flux/functions/KeepFlux.java | 2 +- .../influxdata/flux/functions/LastFlux.java | 4 +- .../influxdata/flux/functions/LimitFlux.java | 4 +- .../influxdata/flux/functions/MapFlux.java | 4 +- .../influxdata/flux/functions/MaxFlux.java | 4 +- .../influxdata/flux/functions/MeanFlux.java | 4 +- .../influxdata/flux/functions/MinFlux.java | 4 +- .../influxdata/flux/functions/PivotFlux.java | 5 +- .../influxdata/flux/functions/RangeFlux.java | 4 +- .../influxdata/flux/functions/RenameFlux.java | 5 +- .../influxdata/flux/functions/SampleFlux.java | 3 +- .../influxdata/flux/functions/SetFlux.java | 4 +- .../influxdata/flux/functions/ShiftFlux.java | 2 +- .../influxdata/flux/functions/SkewFlux.java | 3 +- .../influxdata/flux/functions/SortFlux.java | 4 +- .../influxdata/flux/functions/SpreadFlux.java | 4 +- .../influxdata/flux/functions/StddevFlux.java | 4 +- .../influxdata/flux/functions/SumFlux.java | 3 +- .../influxdata/flux/functions/ToBoolFlux.java | 3 +- .../flux/functions/ToDurationFlux.java | 2 +- .../flux/functions/ToFloatFlux.java | 3 +- .../org/influxdata/flux/functions/ToFlux.java | 213 ++++++++++++++++++ .../influxdata/flux/functions/ToIntFlux.java | 3 +- .../flux/functions/ToStringFlux.java | 4 +- .../influxdata/flux/functions/ToTimeFlux.java | 3 +- .../influxdata/flux/functions/ToUIntFlux.java | 2 +- .../influxdata/flux/functions/WindowFlux.java | 4 +- .../influxdata/flux/functions/YieldFlux.java | 4 +- .../influxdata/flux/functions/ToFluxTest.java | 133 +++++++++++ 45 files changed, 603 insertions(+), 67 deletions(-) create mode 100644 flux-dsl/src/main/java/org/influxdata/flux/functions/ToFlux.java create mode 100644 flux-dsl/src/test/java/org/influxdata/flux/functions/ToFluxTest.java diff --git a/flux-dsl/README.md b/flux-dsl/README.md index 55a5b9f80ce..e2fa7b3e30f 100644 --- a/flux-dsl/README.md +++ b/flux-dsl/README.md @@ -570,6 +570,20 @@ Flux flux = Flux .from("telegraf") .sum(); ``` +### to +The To operation takes data from a stream and writes it to a bucket [[doc](http://bit.ly/flux-spec#to)]. +- `bucket` - The bucket to which data will be written. [string] +- `bucketID` - The ID of the bucket to which data will be written. [string] +- `org` - The organization name of the above bucket. [string] +- `orgID` - The organization ID of the above bucket. [string] +- `host` - The remote host to write to. [string] +- `token` - The authorization token to use when writing to a remote host. [string] +- `timeColumn` - The time column of the output. [string] +- `tagColumns` - The tag columns of the output. **Default:** All columns of type string, excluding all value columns and the `_field` column if present. [array of strings] +- `fieldFn` - Function that takes a record from the input table and returns an object. For each record from the input table fieldFn returns on object that maps output field key to output value. **Default:** `(r) => ({ [r._field]: r._value })` [function(record) object] +```java + +``` ### toBool Convert a value to a bool [[doc](http://bit.ly/flux-spec#tobool)]. diff --git a/flux-dsl/src/main/java/org/influxdata/flux/Flux.java b/flux-dsl/src/main/java/org/influxdata/flux/Flux.java index fddff023bf8..769d0d5d229 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/Flux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/Flux.java @@ -66,6 +66,7 @@ import org.influxdata.flux.functions.ToBoolFlux; import org.influxdata.flux.functions.ToDurationFlux; import org.influxdata.flux.functions.ToFloatFlux; +import org.influxdata.flux.functions.ToFlux; import org.influxdata.flux.functions.ToIntFlux; import org.influxdata.flux.functions.ToStringFlux; import org.influxdata.flux.functions.ToTimeFlux; @@ -121,7 +122,7 @@ *
  • !TODO stateTracking - Not defined in documentation or SPEC
    • *
  • {@link StddevFlux}
    • *
  • {@link SumFlux}
    • - *
  • !TODO - to
    • + *
  • {@link ToFlux}
    • *
  • {@link ToBoolFlux}
    • *
  • {@link ToIntFlux}
    • *
  • {@link ToFloatFlux}
    • @@ -1583,6 +1584,174 @@ public final SumFlux sum(final boolean useStartTime) { return new SumFlux(this).withUseStartTime(useStartTime); } + /** + * To operation takes data from a stream and writes it to a bucket. + * + *

    The parameters had to be defined by:

    + * + * + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to() { + return new ToFlux(this); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @param fieldFn Function that takes a record from the input table and returns an object. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org, + @Nonnull final String fieldFn) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org) + .withFieldFunction(fieldFn); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @param tagColumns The tag columns of the output. + * @param fieldFn Function that takes a record from the input table and returns an object. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org, + @Nonnull final String[] tagColumns, + @Nonnull final String fieldFn) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org) + .withTagColumns(tagColumns) + .withFieldFunction(fieldFn); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @param tagColumns The tag columns of the output. + * @param fieldFn Function that takes a record from the input table and returns an object. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org, + @Nonnull final Collection tagColumns, + @Nonnull final String fieldFn) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org) + .withTagColumns(tagColumns) + .withFieldFunction(fieldFn); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @param host The remote host to write to. + * @param token The authorization token to use when writing to a remote host. + * @param timeColumn The time column of the output. + * @param tagColumns The tag columns of the output. + * @param fieldFn Function that takes a record from the input table and returns an object. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org, + @Nonnull final String host, + @Nonnull final String token, + @Nonnull final String timeColumn, + @Nonnull final String[] tagColumns, + @Nonnull final String fieldFn) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org) + .withHost(host) + .withToken(token) + .withTimeColumn(timeColumn) + .withTagColumns(tagColumns) + .withFieldFunction(fieldFn); + } + + /** + * To operation takes data from a stream and writes it to a bucket. + * + * @param bucket The bucket to which data will be written. + * @param org The organization name of the above bucket. + * @param host The remote host to write to. + * @param token The authorization token to use when writing to a remote host. + * @param timeColumn The time column of the output. + * @param tagColumns The tag columns of the output. + * @param fieldFn Function that takes a record from the input table and returns an object. + * @return {@link ToFlux} + */ + @Nonnull + public final ToFlux to(@Nonnull final String bucket, + @Nonnull final String org, + @Nonnull final String host, + @Nonnull final String token, + @Nonnull final String timeColumn, + @Nonnull final Collection tagColumns, + @Nonnull final String fieldFn) { + + return new ToFlux(this) + .withBucket(bucket) + .withOrg(org) + .withHost(host) + .withToken(token) + .withTimeColumn(timeColumn) + .withTagColumns(tagColumns) + .withFieldFunction(fieldFn); + } + /** * Convert a value to a bool. * diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/CountFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/CountFlux.java index 8c17c205621..f25eea579f2 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/CountFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/CountFlux.java @@ -26,7 +26,7 @@ import org.influxdata.flux.Flux; /** - * count - Counts the number of results. + * Counts the number of results. See SPEC. * *

    Options

    *
      diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/CovarianceFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/CovarianceFlux.java index 698cfa82b7c..ee084dc2797 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/CovarianceFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/CovarianceFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * covariance - * Covariance is an aggregate operation. Covariance computes the covariance between two columns. + * See SPEC. * *

      Options

      *
        diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/CumulativeSumFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/CumulativeSumFlux.java index ae9d2d129b1..7462e4e8301 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/CumulativeSumFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/CumulativeSumFlux.java @@ -28,8 +28,9 @@ import org.influxdata.platform.Arguments; /** - * cumulativeSum - Cumulative sum computes a running sum - * for non null records in the table. The output table schema will be the same as the input table + * Cumulative sum computes a running sum for non null records in the table. + * The output table schema will be the same as the input table. + * See SPEC. * *

        Options

        *
          diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/DerivativeFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/DerivativeFlux.java index 5d041dec0d2..c837bdf7461 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/DerivativeFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/DerivativeFlux.java @@ -29,8 +29,8 @@ import org.influxdata.platform.Arguments; /** - * derivative - * Computes the time based difference between subsequent non null records. + * See SPEC. * *

          Options

          *
            diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/DifferenceFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/DifferenceFlux.java index 51b18fdd8c0..64ba932f629 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/DifferenceFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/DifferenceFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * difference - * Difference computes the difference between subsequent non null records. + * See SPEC. * *

            Options

            *
              diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/DistinctFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/DistinctFlux.java index c52e2d08ffb..81d1047846c 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/DistinctFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/DistinctFlux.java @@ -27,8 +27,8 @@ import org.influxdata.platform.Arguments; /** - * distinct - Distinct - * produces the unique values for a given column. + * Distinct produces the unique values for a given column. + * See SPEC. * *

              Options

              *
                diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/DropFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/DropFlux.java index aab58e842dd..2540b85f099 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/DropFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/DropFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * duplicate - Duplicate will - * duplicate a specified column in a table. + * Drop will exclude specified columns from a table. + * See SPEC. * *

                Options

                *
                  @@ -97,7 +97,6 @@ public DropFlux withColumns(@Nonnull final Collection columns) { return this; } - /** * @param function The function which takes a column name as a parameter and returns a boolean indicating whether * or not the column should be excluded from the resulting table. diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/DuplicateFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/DuplicateFlux.java index ee4ee8eed79..74822231fa8 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/DuplicateFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/DuplicateFlux.java @@ -27,8 +27,8 @@ import org.influxdata.platform.Arguments; /** - * duplicate -Duplicate will - * duplicate a specified column in a table. + * Duplicate will duplicate a specified column in a table. + * See SPEC. * *

                  Options

                  *
                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/FilterFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/FilterFlux.java index 943059b253e..f243b124be1 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/FilterFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/FilterFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * filter - * Filters the results using an expression. + * See SPEC. * *

                    Options

                    *
                      diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/FirstFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/FirstFlux.java index 0d79b45b5bc..ab247a7bfe5 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/FirstFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/FirstFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * first - Returns the first result of - * the query. + * Returns the first result of the query. + * See SPEC. * *

                      Options

                      *
                        diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/FromFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/FromFlux.java index c1798276ecc..577961ebc50 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/FromFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/FromFlux.java @@ -24,8 +24,8 @@ import javax.annotation.Nonnull; /** - * from - starting point - * for all queries. Get data from the specified bucket. + * From produces a stream of tables from the specified bucket. + * See SPEC. * *

                        Options

                        *
                          diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/GroupFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/GroupFlux.java index 592f706986c..541ea30523e 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/GroupFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/GroupFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * group - Groups results by - * a user-specified set of tags. + * Groups results by a user-specified set of tags. + * See SPEC. * *

                          Options

                          *
                            diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/IntegralFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/IntegralFlux.java index 4b417b75cd0..9a2c7fdf5c5 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/IntegralFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/IntegralFlux.java @@ -28,9 +28,9 @@ import org.influxdata.platform.Arguments; /** - * integral - * For each aggregate column, it outputs the area under the curve of non null records. * The curve is defined as function where the domain is the record times and the range is the record values. + * See SPEC. * *

                            Options

                            *
                              diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/JoinFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/JoinFlux.java index 99da836f649..29801534fc0 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/JoinFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/JoinFlux.java @@ -32,8 +32,8 @@ import org.influxdata.platform.Arguments; /** - * join - * Join two time series together on time and the list of on keys. + * See SPEC. * *

                              Options

                              *
                                diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/KeepFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/KeepFlux.java index a4410144e3c..5c40835cc98 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/KeepFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/KeepFlux.java @@ -28,9 +28,9 @@ import org.influxdata.platform.Arguments; /** - * keep - * Keep is the inverse of drop. It will return a table containing only columns that are specified, ignoring all others. * Only columns in the group key that are also specified in keep will be kept in the resulting group key. + * See SPEC. * *

                                Options

                                *
                                  diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/LastFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/LastFlux.java index f28111db2f7..d6e65d15ce8 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/LastFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/LastFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * last - Returns the last result of - * the query. + * Returns the last result of the query. + * See SPEC. * *

                                  Options

                                  *
                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/LimitFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/LimitFlux.java index 2caf005a809..55de3125c83 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/LimitFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/LimitFlux.java @@ -27,8 +27,8 @@ import org.influxdata.platform.Arguments; /** - * limit - Restricts the number of rows - * returned in the results. + * Restricts the number of rows returned in the results. + * See SPEC. * *

                                    Options

                                    *
                                      diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/MapFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/MapFlux.java index 3043ece0555..39581b15039 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/MapFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/MapFlux.java @@ -27,8 +27,8 @@ import org.influxdata.platform.Arguments; /** - * map - Applies a function to - * each row of the table. + * Applies a function to each row of the table. + * See SPEC. * *

                                      Options

                                      *
                                        diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/MaxFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/MaxFlux.java index d3a45916d63..d3d0ef72df5 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/MaxFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/MaxFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * max - Returns the max value - * within the results. + * Returns the max value within the results. + * See SPEC. * *

                                        Options

                                        *
                                          diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/MeanFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/MeanFlux.java index 0504b778d49..03f00de355a 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/MeanFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/MeanFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * mean - Returns the mean of the - * within the results. + * Returns the mean of the within the results. + * See SPEC. * *

                                          Options

                                          *
                                            diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/MinFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/MinFlux.java index db761f20a83..c0e58fdb525 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/MinFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/MinFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * min - Returns the min value within - * the results. + * Returns the min value within the results. + * See SPEC. * *

                                            Options

                                            *
                                              diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/PivotFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/PivotFlux.java index daa0e27dd40..18dee84719b 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/PivotFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/PivotFlux.java @@ -28,8 +28,9 @@ import org.influxdata.platform.Arguments; /** - * pivot - Pivot collects values stored vertically (column-wise) in a table - * and aligns them horizontally (row-wise) into logical sets. + * Pivot collects values stored vertically (column-wise) in a table and aligns them horizontally (row-wise) + * into logical sets. + * See SPEC. * *

                                              Options

                                              *
                                                diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/RangeFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/RangeFlux.java index 01f26c10dc3..fd88cd8f7fa 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/RangeFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/RangeFlux.java @@ -29,8 +29,8 @@ import org.influxdata.platform.Arguments; /** - * range - Filters the results by - * time boundaries. + * Filters the results by time boundaries. + * See SPEC. * *

                                                Options

                                                *
                                                  diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/RenameFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/RenameFlux.java index 4105cc4b273..904316f3792 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/RenameFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/RenameFlux.java @@ -28,9 +28,9 @@ import org.influxdata.platform.Arguments; /** - * rename - * Rename will rename specified columns in a table. If a column is renamed and is part of the group key, - * the column name in the group key will be updated + * the column name in the group key will be updated. + * See SPEC. * *

                                                  Options

                                                  *
                                                    @@ -87,7 +87,6 @@ public RenameFlux withColumns(@Nonnull final Map columns) { return this; } - /** * @param function The function which takes a single string parameter (the old column name) and * returns a string representing the new column name. diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SampleFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SampleFlux.java index 46da71476ba..9b7e5384109 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SampleFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SampleFlux.java @@ -26,7 +26,8 @@ import org.influxdata.flux.Flux; /** - * sample - Sample values from a table. + * Sample values from a table. + * See SPEC. * *

                                                    Options

                                                    *
                                                      diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SetFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SetFlux.java index ffddaba066a..b320acb5c0f 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SetFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SetFlux.java @@ -27,8 +27,8 @@ import org.influxdata.platform.Arguments; /** - * yield - Assigns a static - * value to each record. + * Assigns a static value to each record. + * See SPEC. * *

                                                      Options

                                                      *
                                                        diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ShiftFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ShiftFlux.java index 12db3849a23..0d37e8aa46e 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ShiftFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ShiftFlux.java @@ -29,8 +29,8 @@ import org.influxdata.platform.Arguments; /** - * shift - * Shift add a fixed duration to time columns. + * See SPEC. * *

                                                        Options

                                                        *
                                                          diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SkewFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SkewFlux.java index ee371f9152f..f5d8156b054 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SkewFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SkewFlux.java @@ -26,7 +26,8 @@ import org.influxdata.flux.Flux; /** - * skew - Skew of the results. + * Skew of the results. + * See SPEC. * *

                                                          Options

                                                          *
                                                            diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SortFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SortFlux.java index 335d131bda7..98dedfdbbfe 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SortFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SortFlux.java @@ -28,8 +28,8 @@ import org.influxdata.platform.Arguments; /** - * sort - Sorts the results by the - * specified columns Default sort is ascending. + * Sorts the results by the specified columns Default sort is ascending. + * See SPEC. * *

                                                            Options

                                                            *
                                                              diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SpreadFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SpreadFlux.java index 4bcae3b8364..df66a7b644a 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SpreadFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SpreadFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * spread - Difference between min - * and max values. + * Difference between min and max values. + * See SPEC. * *

                                                              Options

                                                              *
                                                                diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/StddevFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/StddevFlux.java index 403f3b339ff..355d4bda1b1 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/StddevFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/StddevFlux.java @@ -26,8 +26,8 @@ import org.influxdata.flux.Flux; /** - * stddev - Standard Deviation of - * the results. + * Standard Deviation of the results. + * See SPEC. * *

                                                                Options

                                                                *
                                                                  diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/SumFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/SumFlux.java index 84b20a2e309..e52d769f68c 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/SumFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/SumFlux.java @@ -26,7 +26,8 @@ import org.influxdata.flux.Flux; /** - * sum - Sum of the results. + * Sum of the results. + * See SPEC. * *

                                                                  Options

                                                                  *
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToBoolFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToBoolFlux.java index 285e55e933b..fb8cc875421 100644 --- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToBoolFlux.java +++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToBoolFlux.java @@ -26,7 +26,8 @@ import org.influxdata.flux.Flux; /** - * toBool - Convert a value to a bool. + * Convert a value to a bool. + * See SPEC. * *

                                                                    Example

                                                                    * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToDurationFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToDurationFlux.java
index eee34e9357d..b112e174f23 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToDurationFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToDurationFlux.java
@@ -26,8 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toDuration -
  * Convert a value to a duration.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFloatFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFloatFlux.java
index 43025296c4b..880ac62740a 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFloatFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFloatFlux.java
@@ -26,7 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toFloat - Convert a value to a float.
+ * Convert a value to a float.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFlux.java
new file mode 100644
index 00000000000..5e02cdbb198
--- /dev/null
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToFlux.java
@@ -0,0 +1,213 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package org.influxdata.flux.functions;
+
+import java.util.Collection;
+import javax.annotation.Nonnull;
+
+import org.influxdata.flux.Flux;
+import org.influxdata.platform.Arguments;
+
+/**
+ * The To operation takes data from a stream and writes it to a bucket.
+ * See SPEC.
+ *
+ * 
                                                                    Options
                                                                    
+ * 
                                                                      
+ *     
                                                                    • bucket - The bucket to which data will be written. [string]
                                                                      • 
+ *     
                                                                    • bucketID - The bucket to which data will be written. [string]
                                                                      • 
+ *     
                                                                    • org - The organization name of the above bucket. [string]
                                                                      • 
+ *     
                                                                    • orgID - The organization ID of the above bucket. [string]
                                                                      • 
+ *     
                                                                    • host - The remote host to write to. [string]
                                                                      • 
+ *     
                                                                    • token - The authorization token to use when writing to a remote host. [string]
                                                                      • 
+ *     
                                                                    • timeColumn - The time column of the output. [string]
                                                                      • 
+ *     
                                                                    • tagColumns - The tag columns of the output.
                                                                      • 
+ *     
                                                                    • fieldFn - Function that takes a record from the input table and returns an object.
                                                                      • 
+ * 
                                                                    
+ *
+ * @author Jakub Bednar (10/10/2018 07:58)
+ */
+public final class ToFlux extends AbstractParametrizedFlux {
+
+    public ToFlux(@Nonnull final Flux source) {
+        super(source);
+    }
+
+    @Nonnull
+    @Override
+    protected String operatorName() {
+        return "to";
+    }
+
+    /**
+     * @param bucket The bucket to which data will be written.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withBucket(@Nonnull final String bucket) {
+
+        Arguments.checkNonEmpty(bucket, "bucket");
+
+        this.withPropertyValueEscaped("bucket", bucket);
+
+        return this;
+    }
+
+    /**
+     * @param bucketID The ID of the bucket to which data will be written.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withBucketID(@Nonnull final String bucketID) {
+
+        Arguments.checkNonEmpty(bucketID, "bucketID");
+
+        this.withPropertyValueEscaped("bucketID", bucketID);
+
+        return this;
+    }
+
+    /**
+     * @param org The organization name of the above bucket.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withOrg(@Nonnull final String org) {
+
+        Arguments.checkNonEmpty(org, "org");
+
+        this.withPropertyValueEscaped("org", org);
+
+        return this;
+    }
+
+    /**
+     * @param orgID The organization name of the above bucket.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withOrgID(@Nonnull final String orgID) {
+
+        Arguments.checkNonEmpty(orgID, "orgID");
+
+        this.withPropertyValueEscaped("orgID", orgID);
+
+        return this;
+    }
+
+    /**
+     * @param host The remote host to write to.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withHost(@Nonnull final String host) {
+
+        Arguments.checkNonEmpty(host, "host");
+
+        this.withPropertyValueEscaped("host", host);
+
+        return this;
+    }
+
+    /**
+     * @param token The authorization token to use when writing to a remote host.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withToken(@Nonnull final String token) {
+
+        Arguments.checkNonEmpty(token, "token");
+
+        this.withPropertyValueEscaped("token", token);
+
+        return this;
+    }
+
+    /**
+     * @param timeColumn The time column of the output.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withTimeColumn(@Nonnull final String timeColumn) {
+
+        Arguments.checkNonEmpty(timeColumn, "timeColumn");
+
+        this.withPropertyValueEscaped("timeColumn", timeColumn);
+
+        return this;
+    }
+
+    /**
+     * @param tagColumns The tag columns of the output.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withTagColumns(@Nonnull final String[] tagColumns) {
+
+        Arguments.checkNotNull(tagColumns, "tagColumns");
+
+        this.withPropertyValue("tagColumns", tagColumns);
+
+        return this;
+    }
+
+    /**
+     * @param tagColumns The tag columns of the output.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withTagColumns(@Nonnull final Collection tagColumns) {
+
+        Arguments.checkNotNull(tagColumns, "tagColumns");
+
+        this.withPropertyValue("tagColumns", tagColumns);
+
+        return this;
+    }
+
+    /**
+     * @param fieldFn Function that takes a record from the input table and returns an object.
+     * @return this
+     */
+    @Nonnull
+    public ToFlux withFieldFunction(@Nonnull final String fieldFn) {
+
+        Arguments.checkNonEmpty(fieldFn, "fieldFn");
+
+        this.withPropertyValue("fieldFn: (r)", fieldFn);
+
+        return this;
+    }
+
+    @Nonnull
+    @Override
+    protected String propertyDelimiter(@Nonnull final String operatorName) {
+
+        switch (operatorName) {
+            case "fieldFn: (r)":
+                return " => ";
+
+            default:
+                return super.propertyDelimiter(operatorName);
+        }
+    }
+}
diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToIntFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToIntFlux.java
index a704b16c038..160efdeb332 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToIntFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToIntFlux.java
@@ -26,7 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toInt - Convert a value to a int.
+ * Convert a value to a int.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToStringFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToStringFlux.java
index 9dbca99bc67..4903ecb3be3 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToStringFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToStringFlux.java
@@ -26,8 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toString - Convert a value
- * to a string.
+ * Convert a value to a string.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToTimeFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToTimeFlux.java
index 6385cfa814a..9178926ccfc 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToTimeFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToTimeFlux.java
@@ -26,7 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toTime - Convert a value to a time.
+ * Convert a value to a time.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToUIntFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToUIntFlux.java
index 9ec2b4b8186..9fd1a8d1a8f 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/ToUIntFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/ToUIntFlux.java
@@ -26,8 +26,8 @@
 import org.influxdata.flux.Flux;
 
 /**
- * toDuration -
  * Convert a value to a duration.
+ * See SPEC.
  *
  * 
                                                                    Example
                                                                    
  * 
                                                                    diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/WindowFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/WindowFlux.java
index 0f877f22fcd..12b69204e40 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/WindowFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/WindowFlux.java
@@ -29,8 +29,8 @@
 import org.influxdata.platform.Arguments;
 
 /**
- * window - Groups the results by
- * a given time range.
+ * Groups the results by a given time range.
+ * See SPEC.
  *
  * 
                                                                    Options
                                                                    
  * 
                                                                      
diff --git a/flux-dsl/src/main/java/org/influxdata/flux/functions/YieldFlux.java b/flux-dsl/src/main/java/org/influxdata/flux/functions/YieldFlux.java
index 7f22db4a9e4..87408a68fbf 100644
--- a/flux-dsl/src/main/java/org/influxdata/flux/functions/YieldFlux.java
+++ b/flux-dsl/src/main/java/org/influxdata/flux/functions/YieldFlux.java
@@ -27,8 +27,8 @@
 import org.influxdata.platform.Arguments;
 
 /**
- * set - Yield a query results
- * to yielded results.
+ * Yield a query results to yielded results.
+ * See SPEC.
  *
  * 
                                                                      Options
                                                                      
  * 
                                                                        
diff --git a/flux-dsl/src/test/java/org/influxdata/flux/functions/ToFluxTest.java b/flux-dsl/src/test/java/org/influxdata/flux/functions/ToFluxTest.java
new file mode 100644
index 00000000000..a6436136463
--- /dev/null
+++ b/flux-dsl/src/test/java/org/influxdata/flux/functions/ToFluxTest.java
@@ -0,0 +1,133 @@
+/*
+ * The MIT License
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in
+ * all copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+ * THE SOFTWARE.
+ */
+package org.influxdata.flux.functions;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.influxdata.flux.Flux;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.junit.platform.runner.JUnitPlatform;
+import org.junit.runner.RunWith;
+
+/**
+ * @author Jakub Bednar (10/10/2018 09:35)
+ */
+@RunWith(JUnitPlatform.class)
+class ToFluxTest {
+
+    @Test
+    void to() {
+
+        Flux flux = Flux
+                .from("telegraf")
+                .to()
+                    .withBucketID("O1")
+                    .withOrgID("02")
+                    .withHost("example.com")
+                    .withToken("secret")
+                    .withTimeColumn("timestamp")
+                    .withTagColumns(new String[]{"location", "production"})
+                    .withFieldFunction("return {\"hum\": r.hum, \"temp\": r.temp}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucketID: \"O1\", orgID: \"02\", host: \"example.com\", token: \"secret\", timeColumn: \"timestamp\", tagColumns: [\"location\", \"production\"], fieldFn: (r) => return {\"hum\": r.hum, \"temp\": r.temp})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+    @Test
+    void toBucketOrg() {
+
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org");
+
+        Assertions.assertThat(flux.toString())
+                .isEqualToIgnoringWhitespace("from(bucket:\"telegraf\") |> to(bucket: \"my-bucket\", org: \"my-org\")");
+    }
+
+    @Test
+    void toBucketOrgFieldFn() {
+
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org", "return {\"hum\": r.hum}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucket: \"my-bucket\", org: \"my-org\", fieldFn: (r) => return {\"hum\": r.hum})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+    @Test
+    void toBucketOrgTagColumnsArrayFieldFn() {
+
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org", new String[]{"location"},"return {\"hum\": r.hum}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucket: \"my-bucket\", org: \"my-org\", tagColumns: [\"location\"], fieldFn: (r) => return {\"hum\": r.hum})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+    @Test
+    void toBucketOrgTagColumnsCollectionFieldFn() {
+
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org", Arrays.asList("location", "host"),"return {\"hum\": r.hum}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucket: \"my-bucket\", org: \"my-org\", tagColumns: [\"location\", \"host\"], fieldFn: (r) => return {\"hum\": r.hum})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+    @Test
+    void toArray() {
+
+        String[] tagColumns = {"location", "production"};
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org", "example.com", "secret", "timestamp", tagColumns, "return {\"hum\": r.hum, \"temp\": r.temp}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucket: \"my-bucket\", org: \"my-org\", host: \"example.com\", token: \"secret\", timeColumn: \"timestamp\", tagColumns: [\"location\", \"production\"], fieldFn: (r) => return {\"hum\": r.hum, \"temp\": r.temp})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+    @Test
+    void toCollection() {
+
+        List tagColumns = Arrays.asList("location", "production");
+        
+        Flux flux = Flux
+                .from("telegraf")
+                .to("my-bucket", "my-org", "example.com", "secret", "timestamp", tagColumns, "return {\"hum\": r.hum, \"temp\": r.temp}");
+
+        String expected = "from(bucket:\"telegraf\") |> "
+                + "to(bucket: \"my-bucket\", org: \"my-org\", host: \"example.com\", token: \"secret\", timeColumn: \"timestamp\", tagColumns: [\"location\", \"production\"], fieldFn: (r) => return {\"hum\": r.hum, \"temp\": r.temp})";
+        Assertions.assertThat(flux.toString()).isEqualToIgnoringWhitespace(expected);
+    }
+
+}