Vendor-neutral two-level chunking (by size) and placement-class report

README.ko.md

@@ -70,7 +70,8 @@ export ONGGI_STORE='s3://models/onggi?endpoint=s3.lab.example:8333&insecure=1' A
 ```
 
 연구실 노드에서 ATOM Llama-3.1-8B 빌드 묶음 264 GB가 **15.9 GB**로 저장됩니다(push 3분, pull 5분, 원본과 바이트 단위
-동일). 2단계 청크 분할이 새 ATOM 청크를 약 4 KiB 하위 청크로 다시 나눠서, 같은 가중치를 다른 순서로 담은 prefill
+동일). 2단계 청크 분할이 1 MiB 이상인 모든 파일의 새 청크를 형식과 무관하게 크기만 보고 약 4 KiB 하위 청크로 다시 나눠서
+([docs/sub-chunk-policy.md](docs/sub-chunk-policy.md)), 같은 가중치를 다른 순서로 담은 prefill
 변형끼리 중복을 찾습니다(새 데이터 46.8 GB → 24.9 GB). 여기에 청크 단위 압축이 1.6배를 더합니다. `--compress none`,
 `--sub 0`으로 각각 끌 수 있습니다.
 

README.md

@@ -60,7 +60,9 @@ It walks the paths (symlinks are not followed), sorts files by kind (HF source,
 decode/prefill `.rbln`) so source checkpoints come first, chunks files in parallel at content-defined boundaries
 (FastCDC, 16/64/256 KiB by default) and prints, per kind, the logical size, the distinct bytes and the bytes each
 kind adds on top of the earlier ones. `--fixed-block` repeats the measurement with fixed-size blocks for comparison;
-`--pairs N` prints which share of each of the N largest files occurs in each other.
+`--pairs N` prints which share of each of the N largest files occurs in each other; `--classify [--reference
+CHECKPOINT]` adds each file's weight-placement class (referenced, embedded-rearranged, embedded-shifted, new) from
+chunk shares alone.
 
 Store and restore an artifact (only chunks the store lacks are uploaded and are compressed per chunk; every chunk is
 hash-checked on the way back):
@@ -78,7 +80,8 @@ export ONGGI_STORE='s3://models/onggi?endpoint=s3.lab.example:8333&insecure=1' A
 ```
 
 On the lab node, the 264 GB ATOM Llama-3.1-8B build set is stored in **15.9 GB** (push 3 min, pull 5 min,
-byte-identical). Two steps get it there. Two-level chunking re-splits new ATOM chunks into ~4 KiB sub-chunks, so prefill
+byte-identical). Two steps get it there. Two-level chunking re-splits new chunks of every file of at least 1 MiB into ~4 KiB sub-chunks (by size, not by
+format: [docs/sub-chunk-policy.md](docs/sub-chunk-policy.md)), so prefill
 variants that hold the same weights in a different order dedup (24.9 GB new instead of 46.8 GB). Per-chunk compression
 then takes 1.6× off. `--compress none` and `--sub 0` turn the two off. New chunks of sub-chunked files are compressed together in frames
 of up to 255 KiB where that is smaller (#52; `--frame 0` turns it off).

cmd/onggi/csi.go

@@ -76,7 +76,7 @@ func csiNode(args []string, stderr io.Writer) int {
 				return nil, err
 			}
 			return store.OpenWith(ctx, cb, store.Create{Chunker: s.Config.Chunker, Sub: s.Config.Sub,
-				SubKinds: s.Config.SubKinds, PackSize: s.Config.PackSize})
+				SubKinds: s.Config.SubKinds, SubMinFile: s.Config.SubMinFile, PackSize: s.Config.PackSize})
 		}
 		cache, err := openCache(ctx)
 		if err == nil {

cmd/onggi/main.go

@@ -13,6 +13,7 @@ import (
 	"text/tabwriter"
 	"time"
 
+	"github.com/ai-computing/onggi/pkg/chunker"
 	"github.com/ai-computing/onggi/pkg/dedup"
 )
 
@@ -95,6 +96,16 @@ func dedupReport(args []string, stdout, stderr io.Writer) int {
 	fixed := fs.String("fixed-block", "", "also measure fixed-size blocks of this size (e.g. 1MiB)")
 	asJSON := fs.Bool("json", false, "print JSON instead of tables")
 	quiet := fs.Bool("quiet", false, "no progress on stderr")
+	classify := fs.Bool("classify", false, "report each file's weight-placement class from its bytes: referenced, embedded-rearranged, embedded-shifted, new (#74)")
+	var refs []string
+	fs.Func("reference", "with --classify: a file or directory the set may reference, such as the source checkpoint (repeatable; links are followed)", func(v string) error {
+		refs = append(refs, v)
+		return nil
+	})
+	fine := fs.String("fine", "4KiB", "with --classify: average fine chunk size; min and max are a quarter and four times it")
+	refShare := fs.Float64("ref-share", dedup.DefaultRefShare, "with --classify: referenced when at least this share of a file is in the references")
+	gap := fs.Float64("rearranged-gap", dedup.DefaultRearrangeGap, "with --classify: embedded-rearranged when fine chunks find at least this much more of a file elsewhere than coarse chunks")
+	shifted := fs.Float64("shifted-share", dedup.DefaultShiftedShare, "with --classify: embedded-shifted when coarse chunks find at least this share of a file elsewhere")
 	fs.Usage = func() {
 		fmt.Fprintln(stderr, "usage: onggi dedup-report [flags] <path>...")
 		fs.PrintDefaults()
@@ -125,6 +136,16 @@ func dedupReport(args []string, stdout, stderr io.Writer) int {
 	if err == nil {
 		err = o.Params.Validate()
 	}
+	if err == nil && (*classify || len(refs) > 0) {
+		var avg int
+		if avg, err = size(*fine); err == nil {
+			o.Classify = &dedup.Classify{Fine: chunker.Params{Min: avg / 4, Avg: avg, Max: avg * 4}, References: refs,
+				RefShare: *refShare, RearrangedGap: *gap, ShiftedShare: *shifted}
+			if err = o.Classify.Fine.Validate(); err == nil && o.Classify.Fine.Max > o.Params.Min {
+				err = fmt.Errorf("fine chunk max %d exceeds chunk min %d", o.Classify.Fine.Max, o.Params.Min)
+			}
+		}
+	}
 	if err != nil {
 		fmt.Fprintln(stderr, "onggi:", err)
 		return 2
@@ -190,6 +211,19 @@ func printReport(w io.Writer, r *dedup.Report, took time.Duration) {
 		}
 		tw.Flush()
 	}
+	if c := r.Classify; c != nil {
+		fmt.Fprintf(w, "\nplacement (fine chunks %s; referenced >= %.2f in %d reference files, rearranged: fine - coarse >= %.2f, shifted: coarse >= %.2f):\n",
+			human(int64(c.Fine.Avg)), c.RefShare, c.RefFiles, c.RearrangedGap, c.ShiftedShare)
+		tw = tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
+		fmt.Fprintln(tw, "  CLASS\tBYTES\tREF\tCOARSE\tFINE\tPATH")
+		for _, f := range r.Files {
+			if f.Shares != nil {
+				fmt.Fprintf(tw, "  %s\t%s\t%.1f %%\t%.1f %%\t%.1f %%\t%s\n", f.Class, human(f.Bytes), 100*f.Shares.Ref,
+					100*f.Shares.Coarse, 100*f.Shares.Fine, f.Path)
+			}
+		}
+		tw.Flush()
+	}
 	if len(r.Skipped) > 0 {
 		fmt.Fprintf(w, "\nskipped %d symlinks or unreadable entries\n", len(r.Skipped))
 	}

cmd/onggi/main_test.go

@@ -39,3 +39,26 @@ func TestDedupReportRuns(t *testing.T) {
 		t.Fatalf("json output: exit %d\n%s", code, out.String())
 	}
 }
+
+func TestDedupReportClassify(t *testing.T) {
+	dir, ref := t.TempDir(), t.TempDir()
+	data := bytes.Repeat([]byte("onggi "), 200_000)
+	if err := os.WriteFile(filepath.Join(dir, "copy.bin"), data, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(ref, "model.safetensors"), data, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	var out, errb bytes.Buffer
+	if code := dedupReport([]string{"-quiet", "-json", "-reference", ref, dir}, &out, &errb); code != 0 ||
+		!strings.Contains(out.String(), `"class": "referenced"`) || !strings.Contains(out.String(), `"rearranged_gap": 0.25`) {
+		t.Fatalf("classify json: exit %d %s\n%s", code, errb.String(), out.String())
+	}
+	out.Reset()
+	if code := dedupReport([]string{"-quiet", "-classify", dir}, &out, &errb); code != 0 || !strings.Contains(out.String(), "placement") {
+		t.Fatalf("classify table: exit %d\n%s", code, out.String())
+	}
+	if code := dedupReport([]string{"-quiet", "-classify", "-fine", "32KiB", dir}, &out, &errb); code != 2 {
+		t.Fatalf("fine chunks larger than the chunk minimum: exit %d", code)
+	}
+}

cmd/onggi/store.go

@@ -25,7 +25,7 @@ import (
 )
 
 type storeFlags struct {
-	url, durability, min, avg, max, sub, subKinds, packSize, frame *string
+	url, durability, min, avg, max, sub, subKinds, subMinFile, packSize, frame *string
 }
 
 func addStoreFlags(fs *flag.FlagSet) storeFlags {
@@ -36,7 +36,8 @@ func addStoreFlags(fs *flag.FlagSet) storeFlags {
 		avg:        fs.String("avg", "64KiB", "average chunk size (only when creating a store)"),
 		max:        fs.String("max", "256KiB", "maximum chunk size (only when creating a store)"),
 		sub:        fs.String("sub", "4KiB", "average sub-chunk size for two-level chunking, 0 for one level (only when creating a store)"),
-		subKinds:   fs.String("sub-kinds", "atom-prefill,atom-decode,atom-other", "file kinds that get sub-chunks, or all (only when creating a store)"),
+		subKinds:   fs.String("sub-kinds", "", "file kinds (pkg/layout, by name) that get sub-chunks, or all; overrides --sub-min-file (only when creating a store; the behaviour before #74)"),
+		subMinFile: fs.String("sub-min-file", "1MiB", "sub-chunk new chunks of files of at least this size, whatever their name or format (only when creating a store)"),
 		packSize:   fs.String("pack-size", "64MiB", "pack object size (only when creating a store)"),
 		frame:      fs.String("frame", "256KiB", "compress new chunks of sub-chunked files together up to this size, 0 for one by one (only when creating a store)"),
 	}
@@ -95,7 +96,15 @@ func (f storeFlags) open(ctx context.Context) (*store.Store, backend.Refs, error
 			}
 			c.Frame = int64(fr)
 		}
-		if *f.subKinds != "all" {
+		switch *f.subKinds {
+		case "all":
+		case "":
+			mf, err := size(*f.subMinFile)
+			if err != nil {
+				return nil, nil, err
+			}
+			c.SubMinFile = int64(mf)
+		default:
 			for _, k := range strings.Split(*f.subKinds, ",") {
 				c.SubKinds = append(c.SubKinds, layout.Kind(strings.TrimSpace(k)))
 			}
@@ -226,7 +235,7 @@ func pull(args []string, stdout, stderr io.Writer) int {
 		cb, err := bfs.Open(*cacheDir, backend.DurabilityNone)
 		if err == nil {
 			o.Cache, err = store.OpenWith(ctx, cb, store.Create{Chunker: s.Config.Chunker, Sub: s.Config.Sub,
-				SubKinds: s.Config.SubKinds, PackSize: s.Config.PackSize})
+				SubKinds: s.Config.SubKinds, SubMinFile: s.Config.SubMinFile, PackSize: s.Config.PackSize})
 		}
 		if err != nil {
 			fmt.Fprintln(stderr, "onggi: node cache:", err)
@@ -311,9 +320,12 @@ func ls(args []string, stdout, stderr io.Writer) int {
 	fmt.Fprintf(stdout, "\nstore: %d chunks in %d packs, %s; chunker %s/%s/%s; durability %s\n", chunks, packs, human(packBytes),
 		human(int64(c.Min)), human(int64(c.Avg)), human(int64(c.Max)), s.B.Caps().Durability)
 	if sc := s.Config.Sub; sc != nil {
-		kinds := "all kinds"
-		if len(s.Config.SubKinds) > 0 {
-			kinds = fmt.Sprint(s.Config.SubKinds)
+		kinds := "every file"
+		switch {
+		case len(s.Config.SubKinds) > 0:
+			kinds = "kinds " + fmt.Sprint(s.Config.SubKinds)
+		case s.Config.SubMinFile > 0:
+			kinds = "files of at least " + human(s.Config.SubMinFile)
 		}
 		fmt.Fprintf(stdout, "two-level: %d chunks held as sub-chunks of %s/%s/%s (%s)\n", s.Composites(),
 			human(int64(sc.Min)), human(int64(sc.Avg)), human(int64(sc.Max)), kinds)

docs/artifact-formats.md

@@ -120,7 +120,8 @@ All four prefill variants (56.5 GB), compared with the HF source blobs:
 - The share of HF source bytes found is 0.0 % at both sizes.
 
 The variants hold the same bytes rearranged in runs of a few KiB, which finer chunks recover. Their bytes do not occur
-in the HF source at these sizes. The store sub-chunks new ATOM chunks at 4 KiB (two-level chunking).
+in the HF source at these sizes. The store sub-chunks new chunks at 4 KiB (two-level chunking; since #74 for every file of at least 1 MiB, see
+[sub-chunk-policy.md](sub-chunk-policy.md)).
 
 **What remains after 4 KiB sub-chunking** (#9). A decode build shares 87 % of its bytes with a prefill build at 4 KiB,
 so the "second weight layout" seen at 64 KiB is mostly the same bytes in a different order.